ec distribution: collapse chained shard moves to the final destination

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

seaweed-volume/src/server/grpc_server.rs

@@ -53,6 +53,12 @@ fn needle_not_found(needle_id: NeedleId) -> Status {
     Status::not_found(format!("needle not found {}", needle_id.0))
 }
 
+/// Mirrors Go's `isNotLeaderErr`: the master returns raft's NotLeaderError,
+/// which arrives as an Unknown-status message.
+fn is_not_leader_status(e: &Status) -> bool {
+    e.message().contains("Not current leader")
+}
+
 /// Deadline from the caller's `grpc-timeout` header (TimeoutValue +
 /// H|M|S|m|u|n unit, per the gRPC-over-HTTP2 spec). Tonic's GrpcTimeout layer
 /// drops an over-deadline handler future anyway; parsing it here lets a queued
@@ -344,22 +350,46 @@ impl VolumeGrpcService {
         is_readonly: bool,
     ) -> Result<(), Status> {
         // VolumeMarkReadonly mutates raft-replicated master topology, so it must
-        // reach the leader. Prefer the live leader the heartbeat is talking to
-        // (current_master_url), falling back to the static seed before the first
-        // heartbeat — mirrors store_ec.rs and Go's vs.GetMaster(). Sending it to
-        // the static seed fails "not current leader" after any master failover.
-        let master_url = {
-            let live = self.state.current_master_url.read().await.clone();
-            if !live.is_empty() {
-                live
-            } else {
-                self.state.master_url.clone()
-            }
-        };
+        // reach the leader the heartbeat confirmed. On "Not current leader",
+        // ask the seed masters who leads and retry there — mirrors Go's
+        // notifyMasterVolumeReadonly.
+        let master_url = self.state.current_master().await;
         if master_url.is_empty() {
-            return Ok(());
+            if !self.state.has_master {
+                return Ok(());
+            }
+            return Err(Status::unavailable("no master leader known"));
         }
-        let grpc_addr = parse_grpc_address(&master_url).map_err(|e| {
+        match self
+            .volume_mark_readonly_on_master(&master_url, info, is_readonly)
+            .await
+        {
+            Err(e) if is_not_leader_status(&e) => {
+                let leader = super::heartbeat::lookup_leader_from_peers(
+                    &self.state.master_urls,
+                    self.state.outgoing_grpc_tls.as_ref(),
+                )
+                .await
+                .ok_or_else(|| {
+                    Status::unavailable(format!(
+                        "master {} rejected mark-readonly and no peer named a leader",
+                        master_url
+                    ))
+                })?;
+                self.volume_mark_readonly_on_master(&leader, info, is_readonly)
+                    .await
+            }
+            result => result,
+        }
+    }
+
+    async fn volume_mark_readonly_on_master(
+        &self,
+        master_url: &str,
+        info: &MasterVolumeInfo,
+        is_readonly: bool,
+    ) -> Result<(), Status> {
+        let grpc_addr = parse_grpc_address(master_url).map_err(|e| {
             Status::internal(format!("invalid master address {}: {}", master_url, e))
         })?;
         let channel = connect_channel(
@@ -5776,7 +5806,8 @@ impl VolumeCopyJob {
             (&self.dest.data_base_name, self.dest.has_remote_dat);
         let mut preallocate_size: i64 = 0;
         if !has_remote_dat {
-            let grpc_addr = super::heartbeat::to_grpc_address(&state.master_url);
+            let master = state.current_master().await;
+            let grpc_addr = super::heartbeat::to_grpc_address(&master);
             // Race the master-configuration RPC against the caller's
             // response channel: a stalled master (or a slow leader
             // election) would otherwise hold the task and its .note
@@ -5801,7 +5832,7 @@ impl VolumeCopyJob {
                     }
                 }
                 Err(e) => {
-                    tracing::warn!("get master {} configuration: {}", state.master_url, e);
+                    tracing::warn!("get master {} configuration: {}", master, e);
                 }
             }
 

seaweed-volume/src/server/handlers.rs

@@ -603,7 +603,7 @@ async fn batch_delete_file_ids(
             let locations = lookup_volume(
                 &state.http_client,
                 &state.outgoing_http_scheme,
-                &state.master_url,
+                &state.current_master().await,
                 volume_id,
             )
             .await
@@ -675,7 +675,7 @@ async fn do_replicated_request(
     let locations = lookup_volume(
         &state.http_client,
         &state.outgoing_http_scheme,
-        &state.master_url,
+        &state.current_master().await,
         vid,
     )
     .await
@@ -832,7 +832,7 @@ async fn proxy_or_redirect_to_target(
     let locations = match lookup_volume(
         &state.http_client,
         &state.outgoing_http_scheme,
-        &state.master_url,
+        &state.current_master().await,
         vid.0,
     )
     .await
@@ -1238,7 +1238,7 @@ async fn proxy_missing_volume(
     let query_string = uri.query().unwrap_or("").to_string();
     if is_proxied_query(&query_string)
         || state.read_mode == ReadMode::Local
-        || state.master_url.is_empty()
+        || state.current_master().await.is_empty()
     {
         return StatusCode::NOT_FOUND.into_response();
     }
@@ -1282,7 +1282,7 @@ async fn check_download_limit(
         // Go tries proxy to replica once before the wait loop
         // (checkDownloadLimit); it does not retry on each wakeup.
         let should_try_replica =
-            !is_proxied_query(&query_string) && !state.master_url.is_empty() && {
+            !is_proxied_query(&query_string) && !state.current_master().await.is_empty() && {
                 let store = state.store.read().unwrap();
                 store
                     .find_volume(vid)
@@ -2960,7 +2960,9 @@ pub async fn post_handler(
 
     // Replicate to remote volume servers if this volume has replicas.
     // Matches Go's GetWritableRemoteReplications: skip if copy_count == 1.
-    if !is_replicate && write_result.is_ok() && !state.master_url.is_empty() {
+    // No leader-known gate: when the lookup cannot run, the write must fail
+    // rather than be acknowledged on fewer copies than required.
+    if !is_replicate && write_result.is_ok() {
         let needs_replication = {
             let store = state.store.read().unwrap();
             store
@@ -3358,7 +3360,7 @@ pub async fn delete_handler(
     };
 
     let is_replicate = del_query.split('&').any(|p| p == "type=replicate");
-    if !is_replicate && delete_result.is_ok() && !state.master_url.is_empty() {
+    if !is_replicate && delete_result.is_ok() {
         let needs_replication = {
             let store = state.store.read().unwrap();
             store
@@ -4232,7 +4234,7 @@ async fn read_remote_chunk_needle(
     let locations = lookup_volume(
         &state.http_client,
         &state.outgoing_http_scheme,
-        &state.master_url,
+        &state.current_master().await,
         vid.0,
     )
     .await?;
@@ -6671,7 +6673,8 @@ mod tests {
         let s = Arc::get_mut(&mut state).unwrap();
         // Empty self_url makes `url.contains(self_url)` true for every peer.
         s.self_url = "198.51.100.8:8080".to_string();
-        s.master_url = master_url;
+        s.master_url = master_url.clone();
+        s.current_master_url = tokio::sync::RwLock::new(master_url);
         state
     }
 

seaweed-volume/src/server/heartbeat.rs

@@ -155,7 +155,11 @@ pub async fn run_heartbeat_with_state(
                 }
                 PostAction::SleepPulse => {
                     new_leader = None;
-                    tokio::time::sleep(pulse).await;
+                    // The lookup spends part of the retry delay rather than
+                    // adding to it.
+                    let retry_at = tokio::time::Instant::now() + pulse;
+                    refresh_leader_from_peers(&config, &state, pulse).await;
+                    tokio::time::sleep_until(retry_at).await;
                 }
             }
 
@@ -254,10 +258,56 @@ pub async fn try_get_master_configuration(
     Ok(resp.into_inner())
 }
 
+/// Publish the leader the configured masters report, so lookups stop going
+/// to the master the heartbeat just lost while the loop retries. Without an
+/// answer inside `timeout` the published master is cleared: the lost master
+/// is dead or no longer leader, and a follower answers lookups from its own
+/// cache rather than refusing them. Mirrors Go's refreshLeaderFromPeers.
+async fn refresh_leader_from_peers(
+    config: &HeartbeatConfig,
+    state: &Arc<VolumeServerState>,
+    timeout: Duration,
+) {
+    let lookup = lookup_leader_from_peers(
+        &config.master_addresses,
+        state.outgoing_grpc_tls.as_ref(),
+    );
+    match tokio::time::timeout(timeout, lookup).await {
+        Ok(Some(leader)) => publish_current_master(state, &leader).await,
+        _ => state.current_master_url.write().await.clear(),
+    }
+}
+
+/// Ask each seed master for the raft leader and return the first answer, or
+/// None when none replies with one. Mirrors Go's
+/// `operation.LookupRaftLeaderMaster`.
+pub async fn lookup_leader_from_peers(
+    master_addrs: &[String],
+    tls: Option<&super::grpc_client::OutgoingGrpcTlsConfig>,
+) -> Option<String> {
+    for master_addr in master_addrs {
+        let grpc_addr = to_grpc_address(master_addr);
+        if let Ok(resp) = try_get_master_configuration(&grpc_addr, tls).await
+            && !resp.leader.is_empty()
+        {
+            return Some(resp.leader);
+        }
+    }
+    None
+}
+
 fn is_stopping(state: &VolumeServerState) -> bool {
     *state.is_stopping.read().unwrap()
 }
 
+/// Store the master lookups should go to, in the address form it was
+/// announced in — `host:port` or `host:port.grpcPort` — so an explicit gRPC
+/// port survives to the gRPC callers, mirroring Go's setCurrentMaster.
+/// The heartbeat calls this only for a master known to lead.
+async fn publish_current_master(state: &Arc<VolumeServerState>, master: &str) {
+    *state.current_master_url.write().await = master.to_string();
+}
+
 fn duplicate_uuid_retry_delay(retry_count: u32) -> Duration {
     Duration::from_secs((1u64 << retry_count) * 2)
 }
@@ -430,14 +480,6 @@ async fn do_heartbeat(
     let mut response_stream = client.send_heartbeat(stream).await?.into_inner();
 
     info!("Heartbeat stream established with {}", grpc_addr);
-    // Publish the master we're now talking to in canonical http host:port
-    // 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 mut guard = state.current_master_url.write().await;
-        *guard = normalised;
-    }
     if is_stopping(state) {
         state.is_heartbeating.store(false, Ordering::Relaxed);
         send_deregister_heartbeat(config, state, &tx).await;
@@ -451,13 +493,31 @@ async fn do_heartbeat(
     volume_tick.tick().await;
     ec_tick.tick().await;
 
+    let mut confirmed_leader = false;
     loop {
         tokio::select! {
             resp = response_stream.message() => {
                 match resp {
                     Ok(Some(hb_resp)) => {
-                        // Match Go ordering: DuplicatedUuids first, then volume
-                        // options, then leader redirect.
+                        // A reply names the leader or comes from it, so it
+                        // settles where lookups go before anything else on the
+                        // reply is handled: a follower reply carries no other
+                        // fields, and a leader stays the lookup target even
+                        // when it rejects this registration. Mirrors Go.
+                        if !hb_resp.leader.is_empty() {
+                            if super::volume_server::to_http_address(current_master)
+                                != super::volume_server::to_http_address(&hb_resp.leader)
+                            {
+                                publish_current_master(state, &hb_resp.leader).await;
+                                return Ok(Some(hb_resp.leader));
+                            }
+                        } else if !confirmed_leader {
+                            // The leader answers every new stream without
+                            // naming a leader, so this peer is the one lookups
+                            // go to.
+                            confirmed_leader = true;
+                            publish_current_master(state, current_master).await;
+                        }
                         if !hb_resp.duplicated_uuids.is_empty() {
                             let uuids = hb_resp.duplicated_uuids.clone();
                             let duplicate_dirs = off_runtime(config, state, move |_, state| {
@@ -502,11 +562,6 @@ async fn do_heartbeat(
                         if metrics_changed {
                             state.metrics_notify.notify_waiters();
                         }
-                        // Match Go: only redirect if leader is non-empty AND
-                        // different from the current master we're connected to.
-                        if !hb_resp.leader.is_empty() && current_master != hb_resp.leader {
-                            return Ok(Some(hb_resp.leader));
-                        }
                     }
                     Ok(None) => return Ok(None),
                     Err(e) => return Err(Box::new(e)),

seaweed-volume/src/server/store_ec.rs

@@ -1139,16 +1139,9 @@ async fn cached_lookup_ec_shard_locations(
     state: &Arc<VolumeServerState>,
     vid: VolumeId,
 ) -> io::Result<HashMap<ShardId, Vec<String>>> {
-    let master = {
-        let live = state.current_master_url.read().await.clone();
-        if !live.is_empty() {
-            live
-        } else {
-            state.master_url.clone()
-        }
-    };
+    let master = state.current_master().await;
     if master.is_empty() {
-        return Err(io::Error::other("no master configured for ec shard lookup"));
+        return Err(io::Error::other("no master leader known for ec shard lookup"));
     }
 
     let grpc_addr =

seaweed-volume/src/server/volume_server.rs

@@ -86,7 +86,8 @@ pub struct VolumeServerState {
     /// If true, FetchAndWriteNeedle skips remote S3 endpoint validation,
     /// allowing arbitrary (incl. loopback / link-local / metadata) hosts.
     pub allow_untrusted_remote_endpoints: bool,
-    /// First master address for volume lookups (e.g., "localhost:9333").
+    /// First configured master address (e.g., "localhost:9333"). A seed, not a
+    /// lookup target: lookups go to `current_master_url`.
     pub master_url: String,
     /// Seed master addresses for UI rendering.
     pub master_urls: Vec<String>,
@@ -94,10 +95,12 @@ pub struct VolumeServerState {
     /// Built once at construction so Ping admission stays O(1). Mirrors
     /// Go's `seedMasterSet` on `VolumeServer`.
     pub seed_master_set: std::collections::HashSet<String>,
-    /// Current master this server is heartbeating with, in canonical http
-    /// `host:port` form. Empty when no heartbeat connection is active. The
-    /// heartbeat goroutine writes; admission reads — the lock keeps them
-    /// from racing on a leader change. Mirrors Go's `currentMaster` plus
+    /// Master this server takes as leader, in its announced
+    /// `host:port[.grpcPort]` form — callers pick http or gRPC form per use.
+    /// Empty until the heartbeat confirms one — a peer that answered a
+    /// stream without naming a leader, or the leader a follower named — and
+    /// cleared again when no master reports one. The heartbeat writes;
+    /// admission and lookups read. Mirrors Go's `currentMaster` plus
     /// `currentMasterLock`.
     pub current_master_url: tokio::sync::RwLock<String>,
     /// This server's own address (ip:port) for filtering self from lookup results.
@@ -140,6 +143,13 @@ pub struct VolumeServerState {
 }
 
 impl VolumeServerState {
+    /// The master lookups should go to: the leader the heartbeat confirmed,
+    /// or empty while none is known, in its announced `host:port[.grpcPort]`
+    /// form. Mirrors Go's `vs.getCurrentMaster()`.
+    pub async fn current_master(&self) -> String {
+        self.current_master_url.read().await.clone()
+    }
+
     /// Check if the server is in maintenance mode; return gRPC error if so.
     pub fn check_maintenance(&self) -> Result<(), tonic::Status> {
         if self.maintenance.load(Ordering::Relaxed) {
@@ -209,7 +219,7 @@ impl VolumeServerState {
             return false;
         }
         let current = self.current_master_url.read().await.clone();
-        if !current.is_empty() && current == key {
+        if !current.is_empty() && to_http_address(&current) == key {
             return true;
         }
         self.seed_master_set.contains(&key)

weed/s3api/cors/cors.go

@@ -235,13 +235,15 @@ func buildPreflightResponse(rule *CORSRule, corsReq *CORSRequest) *CORSResponse
 
 		// Set exposed headers
 		if len(rule.ExposeHeaders) > 0 {
-			response.ExposeHeaders = strings.Join(rule.ExposeHeaders, ", ")
+			response.ExposeHeaders = formatExposeHeaders(rule.ExposeHeaders)
 		}
 
 		// Set max age
 		if rule.MaxAgeSeconds != nil {
 			response.MaxAge = strconv.Itoa(*rule.MaxAgeSeconds)
 		}
+
+		response.AllowCredentials = true
 	}
 
 	return response
@@ -250,7 +252,8 @@ func buildPreflightResponse(rule *CORSRule, corsReq *CORSRequest) *CORSResponse
 // buildResponse builds a CORS response from a matching rule
 func buildResponse(rule *CORSRule, corsReq *CORSRequest) *CORSResponse {
 	response := &CORSResponse{
-		AllowOrigin: corsReq.Origin,
+		AllowOrigin:      corsReq.Origin,
+		AllowCredentials: true,
 	}
 
 	// Set allowed methods
@@ -263,7 +266,7 @@ func buildResponse(rule *CORSRule, corsReq *CORSRequest) *CORSResponse {
 
 	// Set expose headers
 	if len(rule.ExposeHeaders) > 0 {
-		response.ExposeHeaders = strings.Join(rule.ExposeHeaders, ", ")
+		response.ExposeHeaders = formatExposeHeaders(rule.ExposeHeaders)
 	}
 
 	// Set max age
@@ -274,6 +277,47 @@ func buildResponse(rule *CORSRule, corsReq *CORSRequest) *CORSResponse {
 	return response
 }
 
+// defaultExposeHeaders are emitted alongside "*" so credentialed CORS requests
+// can read standard response headers ("*" is treated literally when credentials
+// are included, per the Fetch spec).
+var defaultExposeHeaders = []string{
+	"ETag",
+	"Content-Length",
+	"Content-Type",
+	"Last-Modified",
+	"Content-Range",
+	"x-amz-request-id",
+	"x-amz-version-id",
+	"x-amz-tagging-count",
+}
+
+// formatExposeHeaders formats the exposed headers slice into a comma-separated string.
+func formatExposeHeaders(exposeHeaders []string) string {
+	var headers []string
+	hasWildcard := false
+	seen := map[string]bool{}
+	for _, h := range exposeHeaders {
+		if h == "*" {
+			hasWildcard = true
+			continue
+		}
+		if lower := strings.ToLower(h); !seen[lower] {
+			seen[lower] = true
+			headers = append(headers, h)
+		}
+	}
+	if hasWildcard {
+		for _, h := range defaultExposeHeaders {
+			if lower := strings.ToLower(h); !seen[lower] {
+				seen[lower] = true
+				headers = append(headers, h)
+			}
+		}
+		headers = append(headers, "*")
+	}
+	return strings.Join(headers, ", ")
+}
+
 // Helper functions
 
 // matchesOrigin checks if the request origin matches any allowed origin

weed/s3api/cors/cors_test.go

@@ -405,11 +405,12 @@ func TestEvaluateRequest(t *testing.T) {
 				Method: "GET",
 			},
 			want: &CORSResponse{
-				AllowOrigin:   "http://example.com",
-				AllowMethods:  "GET, POST",
-				AllowHeaders:  "Content-Type",
-				ExposeHeaders: "ETag",
-				MaxAge:        "3600",
+				AllowOrigin:      "http://example.com",
+				AllowMethods:     "GET, POST",
+				AllowHeaders:     "Content-Type",
+				ExposeHeaders:    "ETag",
+				MaxAge:           "3600",
+				AllowCredentials: true,
 			},
 			wantErr: false,
 		},
@@ -421,8 +422,9 @@ func TestEvaluateRequest(t *testing.T) {
 				Method: "PUT",
 			},
 			want: &CORSResponse{
-				AllowOrigin:  "http://other.com",
-				AllowMethods: "PUT",
+				AllowOrigin:      "http://other.com",
+				AllowMethods:     "PUT",
+				AllowCredentials: true,
 			},
 			wantErr: false,
 		},
@@ -447,11 +449,12 @@ func TestEvaluateRequest(t *testing.T) {
 				AccessControlRequestHeaders: []string{"Content-Type"},
 			},
 			want: &CORSResponse{
-				AllowOrigin:   "http://example.com",
-				AllowMethods:  "GET, POST",
-				AllowHeaders:  "Content-Type",
-				ExposeHeaders: "ETag",
-				MaxAge:        "3600",
+				AllowOrigin:      "http://example.com",
+				AllowMethods:     "GET, POST",
+				AllowHeaders:     "Content-Type",
+				ExposeHeaders:    "ETag",
+				MaxAge:           "3600",
+				AllowCredentials: true,
 			},
 			wantErr: false,
 		},
@@ -467,7 +470,30 @@ func TestEvaluateRequest(t *testing.T) {
 			},
 			want: &CORSResponse{
 				AllowOrigin: "http://example.com",
-				// No AllowMethods or AllowHeaders because the requested header is forbidden
+				// No AllowMethods, AllowHeaders, or AllowCredentials because the requested header is forbidden
+			},
+			wantErr: false,
+		},
+		{
+			name: "wildcard expose headers expands for credentialed responses",
+			config: &CORSConfiguration{
+				CORSRules: []CORSRule{
+					{
+						AllowedOrigins: []string{"http://example.com"},
+						AllowedMethods: []string{"GET"},
+						ExposeHeaders:  []string{"*"},
+					},
+				},
+			},
+			corsReq: &CORSRequest{
+				Origin: "http://example.com",
+				Method: "GET",
+			},
+			want: &CORSResponse{
+				AllowOrigin:      "http://example.com",
+				AllowMethods:     "GET",
+				ExposeHeaders:    "ETag, Content-Length, Content-Type, Last-Modified, Content-Range, x-amz-request-id, x-amz-version-id, x-amz-tagging-count, *",
+				AllowCredentials: true,
 			},
 			wantErr: false,
 		},
@@ -511,18 +537,20 @@ func TestApplyHeaders(t *testing.T) {
 		{
 			name: "complete response",
 			corsResp: &CORSResponse{
-				AllowOrigin:   "http://example.com",
-				AllowMethods:  "GET, POST",
-				AllowHeaders:  "Content-Type",
-				ExposeHeaders: "ETag",
-				MaxAge:        "3600",
+				AllowOrigin:      "http://example.com",
+				AllowMethods:     "GET, POST",
+				AllowHeaders:     "Content-Type",
+				ExposeHeaders:    "ETag",
+				MaxAge:           "3600",
+				AllowCredentials: true,
 			},
 			want: map[string]string{
-				"Access-Control-Allow-Origin":   "http://example.com",
-				"Access-Control-Allow-Methods":  "GET, POST",
-				"Access-Control-Allow-Headers":  "Content-Type",
-				"Access-Control-Expose-Headers": "ETag",
-				"Access-Control-Max-Age":        "3600",
+				"Access-Control-Allow-Origin":      "http://example.com",
+				"Access-Control-Allow-Methods":     "GET, POST",
+				"Access-Control-Allow-Headers":     "Content-Type",
+				"Access-Control-Expose-Headers":    "ETag",
+				"Access-Control-Max-Age":           "3600",
+				"Access-Control-Allow-Credentials": "true",
 			},
 		},
 		{

weed/s3api/s3api_object_handlers.go

@@ -181,8 +181,10 @@ func (s3a *S3ApiServer) parseAndValidateRange(w http.ResponseWriter, r *http.Req
 		return 0, totalSize, false, nil
 	}
 
-	// Empty directory: 404. A file promoted to a directory keeps its data and stays retrievable.
-	if entry.IsDirectory && totalSize == 0 {
+	// Empty directory: 404. A file promoted to a directory keeps its data and
+	// stays retrievable, and a stored directory-key marker is an object, not a
+	// missing key — an unsatisfiable range answers 416.
+	if entry.IsDirectory && totalSize == 0 && !entry.IsDirectoryKeyObject() {
 		s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
 		return 0, 0, false, newStreamErrorWithResponse(fmt.Errorf("directory object %s/%s cannot be retrieved", bucket, object))
 	}
@@ -533,6 +535,24 @@ func (s3a *S3ApiServer) serveDirectoryContent(w http.ResponseWriter, r *http.Req
 	if r.Method == http.MethodHead && len(entry.Chunks) > 0 {
 		contentLength = int64(entry.Attributes.FileSize)
 	}
+	statusCode := http.StatusOK
+	if r.Header.Get("Range") != "" && requestedPartNumber(r) == 0 {
+		// A ranged request answers like the object path does: 206 with its
+		// Content-Range and the range's length, served on GET, reported on HEAD.
+		totalSize := contentLength
+		offset, size, isRangeRequest, rangeErr := s3a.parseAndValidateRange(w, r, entry, totalSize, bucket, object)
+		if rangeErr != nil {
+			return
+		}
+		if isRangeRequest {
+			contentLength = size
+			w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, offset+size-1, totalSize))
+			statusCode = http.StatusPartialContent
+			if r.Method != http.MethodHead {
+				content = content[offset : offset+size]
+			}
+		}
+	}
 	w.Header().Set("Content-Length", strconv.FormatInt(contentLength, 10))
 
 	// Set last modified
@@ -545,12 +565,12 @@ func (s3a *S3ApiServer) serveDirectoryContent(w http.ResponseWriter, r *http.Req
 
 	// For HEAD requests, don't write body
 	if r.Method == http.MethodHead {
-		w.WriteHeader(http.StatusOK)
+		w.WriteHeader(statusCode)
 		return
 	}
 
 	// Write content
-	w.WriteHeader(http.StatusOK)
+	w.WriteHeader(statusCode)
 	if len(content) > 0 {
 		if _, err := w.Write(content); err != nil {
 			glog.Errorf("serveDirectoryContent: failed to write response: %v", err)
@@ -2745,6 +2765,17 @@ func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request
 		w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", startOffset, endOffset, totalSize))
 		statusCode = http.StatusPartialContent
 		glog.V(3).Infof("HeadObject: part %d is bytes %d-%d/%d", partNumber, startOffset, endOffset, totalSize)
+	} else if r.Header.Get("Range") != "" {
+		// A ranged HEAD reports the range like GET does: 206 with its Content-Range and length
+		offset, size, isRangeRequest, rangeErr := s3a.parseAndValidateRange(w, r, objectEntryForSSE, totalSize, bucket, object)
+		if rangeErr != nil {
+			return
+		}
+		if isRangeRequest {
+			responseSize = size
+			w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, offset+size-1, totalSize))
+			statusCode = http.StatusPartialContent
+		}
 	}
 
 	s3a.setResponseHeaders(w, r, objectEntryForSSE, responseSize)

weed/s3api/s3api_object_handlers_copy.go

@@ -2,17 +2,21 @@ package s3api
 
 import (
 	"bytes"
+	"cmp"
 	"context"
+	"crypto/md5"
 	"crypto/rand"
 	"encoding/base64"
 	"encoding/xml"
 	"errors"
 	"fmt"
+	"hash"
 	"io"
 	"net/http"
 	"net/url"
 	"os"
 	"reflect"
+	"slices"
 	"strconv"
 	"strings"
 	"time"
@@ -1146,7 +1150,9 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
 		return
 	}
 
-	if uploadEntryHasSSE(uploadEntry) || sourceEntryIsEncrypted(entry) || uploadEntryHasChecksum(uploadEntry) {
+	// Inline-Content sources have no chunks for the raw-byte path to copy;
+	// the plaintext stream serves their range instead.
+	if uploadEntryHasSSE(uploadEntry) || sourceEntryIsEncrypted(entry) || uploadEntryHasChecksum(uploadEntry) || len(entry.Content) > 0 {
 		etag, sseMetadata, errCode := s3a.copyObjectPartViaReencryption(r, entry, startOffset, endOffset, dstBucket, dstObject, uploadID, partID, uploadEntry)
 		if errCode != s3err.ErrNone {
 			s3err.WriteErrorResponse(w, r, errCode)
@@ -1204,15 +1210,24 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
 		// For zero-size files or invalid ranges, create an empty part with size 0
 		dstEntry.Attributes.FileSize = 0
 		dstEntry.Chunks = nil
+		dstEntry.Attributes.Md5 = md5.New().Sum(nil)
 	} else {
 		// Copy chunks that overlap with the range
-		dstChunks, err := s3a.copyChunksForRange(entry, startOffset, endOffset, dstAssignPath)
+		dstChunks, partMd5, err := s3a.copyChunksForRange(entry, startOffset, endOffset, dstAssignPath)
 		if err != nil {
 			glog.Errorf("CopyObjectPartHandler copy chunks error: %v", err)
 			s3err.WriteErrorResponse(w, r, s3a.mapCopyErrorToS3Error(err))
 			return
 		}
 		dstEntry.Chunks = dstChunks
+		dstEntry.Attributes.Md5 = partMd5
+	}
+
+	// Mirror putToFiler: the part entry carries the MD5 of its bytes so
+	// CopyPartResult, ListParts and complete's composite ETag all describe
+	// the copied range rather than the source chunks.
+	if len(dstEntry.Attributes.Md5) > 0 {
+		dstEntry.Extended[s3_constants.ExtETagKey] = []byte(fmt.Sprintf("%x", dstEntry.Attributes.Md5))
 	}
 
 	// Save the part entry to the multipart uploads folder
@@ -1479,7 +1494,7 @@ func (s3a *S3ApiServer) copySingleChunk(chunk *filer_pb.FileChunk, dstPath strin
 	// the source volume for compressed bytes, so a gzipped chunk is
 	// forwarded to the destination without anyone having to decompress.
 	if canStreamCopyChunk(chunk) {
-		if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, 0, int64(chunk.Size), true /*isFullChunk*/, assignResult); err != nil {
+		if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, 0, int64(chunk.Size), true /*isFullChunk*/, assignResult, nil); err != nil {
 			return nil, fmt.Errorf("stream chunk: %w", err)
 		}
 		return dstChunk, nil
@@ -1534,9 +1549,22 @@ func (s3a *S3ApiServer) copySingleChunkForRange(originalChunk, rangeChunk *filer
 	// part-size = chunk-size assemble pattern hits this branch.
 	if canStreamCopyChunk(originalChunk) {
 		isFullChunk := offsetInChunk == 0 && rangeChunk.Size == originalChunk.Size
-		if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, offsetInChunk, int64(rangeChunk.Size), isFullChunk, assignResult); err != nil {
+		// A partial range stores different bytes than the source chunk, so
+		// its ETag has to be recomputed; a whole-chunk copy reuses it.
+		var digest hash.Hash
+		if !isFullChunk {
+			digest = md5.New()
+		}
+		if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, offsetInChunk, int64(rangeChunk.Size), isFullChunk, assignResult, digest); err != nil {
 			return nil, fmt.Errorf("stream chunk range: %w", err)
 		}
+		if digest != nil {
+			dstChunk.ETag = base64.StdEncoding.EncodeToString(digest.Sum(nil))
+			// The range fetch returns raw bytes and the destination stores
+			// them uncompressed, so the source chunk's gzip flag must not
+			// carry over onto a partial copy.
+			dstChunk.IsCompressed = false
+		}
 		return dstChunk, nil
 	}
 
@@ -1609,8 +1637,16 @@ func parseRangeHeader(rangeHeader string, fileSize int64) (startOffset, endOffse
 	return startOffset, endOffset, nil
 }
 
-// copyChunksForRange copies chunks that overlap with the specified range
-func (s3a *S3ApiServer) copyChunksForRange(entry *filer_pb.Entry, startOffset, endOffset int64, dstPath string) ([]*filer_pb.FileChunk, error) {
+// partMd5Result carries the whole-part MD5 the copiedRangeMd5 goroutine
+// computes beside the per-chunk copies.
+type partMd5Result struct {
+	sum []byte
+	err error
+}
+
+// copyChunksForRange copies chunks that overlap with the specified range and
+// returns the MD5 of the copied bytes for the part entry's Attributes.Md5.
+func (s3a *S3ApiServer) copyChunksForRange(entry *filer_pb.Entry, startOffset, endOffset int64, dstPath string) ([]*filer_pb.FileChunk, []byte, error) {
 	var relevantChunks []*filer_pb.FileChunk
 	var originalChunks []*filer_pb.FileChunk
 
@@ -1641,6 +1677,36 @@ func (s3a *S3ApiServer) copyChunksForRange(entry *filer_pb.Entry, startOffset, e
 		}
 	}
 
+	// The part's Attributes.Md5 must cover the copied range in byte order,
+	// and md5 cannot combine per-chunk digests. A single chunk's own digest
+	// is the answer only when it supplies the whole range; anything else —
+	// several chunks, or gaps a sparse source zero-fills — needs an ordered
+	// pass the concurrent copies cannot feed, so a second stream of the
+	// source range hashes beside them.
+	partSize := endOffset - startOffset + 1
+	singleChunkCovers := len(relevantChunks) == 1 &&
+		relevantChunks[0].Offset == 0 && int64(relevantChunks[0].Size) == partSize
+
+	md5Ctx, cancelMd5 := context.WithCancel(context.Background())
+	defer cancelMd5()
+
+	var partMd5 []byte
+	var partMd5Ch chan partMd5Result
+	switch {
+	case singleChunkCovers:
+		// resolved below from the copied chunk's own digest
+	case len(entry.GetChunks()) == 0:
+		// no chunks to read at all: the part is partSize zero bytes (the
+		// stream helper returns empty for a chunkless entry, not zero-fill)
+		partMd5 = md5Zeros(partSize)
+	default:
+		partMd5Ch = make(chan partMd5Result, 1)
+		go func() {
+			sum, err := s3a.copiedRangeMd5(md5Ctx, entry, startOffset, endOffset)
+			partMd5Ch <- partMd5Result{sum, err}
+		}()
+	}
+
 	// Copy the relevant chunks using a specialized method for range copies
 	dstChunks := make([]*filer_pb.FileChunk, len(relevantChunks))
 	const defaultChunkCopyConcurrency = 4
@@ -1664,11 +1730,79 @@ func (s3a *S3ApiServer) copyChunksForRange(entry *filer_pb.Entry, startOffset, e
 	// Wait for all operations to complete and check for errors
 	for i := 0; i < len(relevantChunks); i++ {
 		if err := <-errChan; err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 	}
 
-	return dstChunks, nil
+	// Reads resolve overlapping chunks by ModifiedTsNs, and the concurrent
+	// copies stamped arbitrary finish times: rank the destination chunks by
+	// the source's timestamps so the part resolves overlaps the way the
+	// source did (and the way copiedRangeMd5 hashed it).
+	if len(dstChunks) > 1 {
+		order := make([]int, len(dstChunks))
+		for i := range order {
+			order[i] = i
+		}
+		slices.SortStableFunc(order, func(a, b int) int {
+			return cmp.Compare(originalChunks[a].ModifiedTsNs, originalChunks[b].ModifiedTsNs)
+		})
+		now := time.Now().UnixNano()
+		for rank, i := range order {
+			dstChunks[i].ModifiedTsNs = now + int64(rank)
+		}
+	}
+
+	if partMd5Ch != nil {
+		if res := <-partMd5Ch; res.err != nil {
+			return nil, nil, fmt.Errorf("md5 of copied range: %w", res.err)
+		} else {
+			partMd5 = res.sum
+		}
+	} else if singleChunkCovers {
+		if partMd5 = util.Base64Md5ToBytes(dstChunks[0].ETag); len(partMd5) != md5.Size {
+			// TUS and mount writes leave chunks without an ETag; hash the
+			// copied range when the inherited digest is missing or malformed.
+			sum, err := s3a.copiedRangeMd5(md5Ctx, entry, startOffset, endOffset)
+			if err != nil {
+				return nil, nil, fmt.Errorf("md5 of copied range: %w", err)
+			}
+			partMd5 = sum
+		}
+	}
+
+	return dstChunks, partMd5, nil
+}
+
+// md5Zeros hashes size zero bytes without a volume read — a copied range
+// that overlaps no source chunk assembles to all zeros.
+func md5Zeros(size int64) []byte {
+	h := md5.New()
+	var zeros [1 << 16]byte
+	for size > 0 {
+		n := min(size, int64(len(zeros)))
+		h.Write(zeros[:n])
+		size -= n
+	}
+	return h.Sum(nil)
+}
+
+// copiedRangeMd5 streams the source range once and hashes its plaintext —
+// exactly the bytes the non-SSE part copy assembles on the destination.
+func (s3a *S3ApiServer) copiedRangeMd5(ctx context.Context, entry *filer_pb.Entry, startOffset, endOffset int64) ([]byte, error) {
+	size := endOffset - startOffset + 1
+	// The copy walks the chunks, so the hash must too: on an entry that also
+	// carries inline Content the stream helper would prefer Content instead.
+	chunked := &filer_pb.Entry{Chunks: entry.GetChunks()}
+	reader, err := s3a.getEncryptedStreamFromVolumes(ctx, chunked, startOffset, size)
+	if err != nil {
+		return nil, err
+	}
+	defer reader.Close()
+	h := md5.New()
+	if _, err := io.Copy(h, reader); err != nil {
+		return nil, err
+	}
+	return h.Sum(nil), nil
 }
 
 // Helper methods for copy operations to avoid code duplication

weed/s3api/s3api_object_handlers_copy_bench_test.go

@@ -160,7 +160,7 @@ func BenchmarkCopyChunk_Streamed(b *testing.B) {
 			for i := 0; i < b.N; i++ {
 				if err := s3a.streamCopyChunkRange(context.Background(),
 					env.srcSrv.URL, env.assign.FileId, 0, int64(size),
-					true /*isFullChunk*/, env.assign); err != nil {
+					true /*isFullChunk*/, env.assign, nil); err != nil {
 					b.Fatalf("stream: %v", err)
 				}
 			}

weed/s3api/s3api_object_handlers_copy_etag_test.go

@@ -1,6 +1,11 @@
 package s3api
 
 import (
+	"bytes"
+	"crypto/md5"
+	"encoding/base64"
+	"fmt"
+	"net/http"
 	"net/http/httptest"
 	"strings"
 	"testing"
@@ -9,6 +14,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
 )
 
 func TestCopyEntryETagPrefersStoredExtendedETag(t *testing.T) {
@@ -145,3 +151,292 @@ func TestParseRangeHeaderRejectsUnsatisfiableRange(t *testing.T) {
 		})
 	}
 }
+
+// A part copied with x-amz-copy-source-range must report the MD5 of the copied
+// bytes: per destination chunk for a partial-chunk copy, and the whole-part MD5
+// the part entry's Attributes.Md5 carries for CopyPartResult / ListParts /
+// complete's composite ETag.
+func TestCopyChunksForRangeCopiedETags(t *testing.T) {
+	volume := startFakeVolumeServer(t)
+	filerImpl := &ambiguousPutFiler{volume: volume, entries: map[string]*filer_pb.Entry{}}
+	s3a := newPutTestServer(t, startFakeFiler(t, filerImpl))
+	util_http.InitGlobalHttpClient()
+
+	const chunkSize = 4 << 20
+	payload := make([]byte, 2*chunkSize)
+	for i := range payload {
+		payload[i] = byte(i*31 + 7)
+	}
+	b64md5 := func(data []byte) string {
+		sum := md5.Sum(data)
+		return base64.StdEncoding.EncodeToString(sum[:])
+	}
+	fid0 := fmt.Sprintf("3,%016x%08x", 101, 202)
+	fid1 := fmt.Sprintf("3,%016x%08x", 103, 204)
+	volume.stored[fid0] = payload[:chunkSize]
+	volume.stored[fid1] = payload[chunkSize:]
+	entry := &filer_pb.Entry{
+		Name: "src",
+		Attributes: &filer_pb.FuseAttributes{
+			FileSize: uint64(len(payload)),
+		},
+		Chunks: []*filer_pb.FileChunk{
+			{FileId: fid0, Offset: 0, Size: chunkSize, ETag: b64md5(payload[:chunkSize])},
+			{FileId: fid1, Offset: chunkSize, Size: chunkSize, ETag: b64md5(payload[chunkSize:])},
+		},
+	}
+
+	// A range spanning the first chunk fully plus 1 MiB of the second.
+	multiEnd := int64(chunkSize + 1<<20 - 1)
+	dstChunks, partMd5, err := s3a.copyChunksForRange(entry, 0, multiEnd, "/dst")
+	if err != nil {
+		t.Fatalf("copyChunksForRange: %v", err)
+	}
+	if len(dstChunks) != 2 {
+		t.Fatalf("copyChunksForRange returned %d chunks, want 2", len(dstChunks))
+	}
+	// A whole-chunk copy forwards the stored bytes, so the source ETag stays valid.
+	if got, want := dstChunks[0].ETag, entry.Chunks[0].ETag; got != want {
+		t.Errorf("whole-chunk copy ETag = %q, want source chunk ETag %q", got, want)
+	}
+	// A partial copy stores only the range bytes; its ETag must be their MD5.
+	if got, want := dstChunks[1].ETag, b64md5(payload[chunkSize:multiEnd+1]); got != want {
+		t.Errorf("partial-chunk copy ETag = %q, want copied-bytes MD5 %q", got, want)
+	}
+	// The copied bytes landed on the destination volume (inside the multipart framing).
+	if stored := volume.stored[dstChunks[1].GetFileIdString()]; !bytes.Contains(stored, payload[chunkSize:multiEnd+1]) {
+		t.Error("destination volume does not hold the copied range bytes")
+	}
+	// The whole-part MD5 hashes every copied byte once, in range order.
+	wantPartMd5 := md5.Sum(payload[:multiEnd+1])
+	if !bytes.Equal(partMd5, wantPartMd5[:]) {
+		t.Errorf("part Md5 = %x, want %x", partMd5, wantPartMd5)
+	}
+	partEntry := &filer_pb.Entry{
+		Attributes: &filer_pb.FuseAttributes{Md5: partMd5},
+		Chunks:     dstChunks,
+	}
+	if got, want := copyEntryETag(partEntry), fmt.Sprintf("%x", wantPartMd5); got != want {
+		t.Errorf("copyEntryETag = %q, want %q", got, want)
+	}
+
+	// A range inside one chunk: the part's ETag is the MD5 of those bytes.
+	innerChunks, innerMd5, err := s3a.copyChunksForRange(entry, 100, 199, "/dst")
+	if err != nil {
+		t.Fatalf("copyChunksForRange inner range: %v", err)
+	}
+	if len(innerChunks) != 1 {
+		t.Fatalf("inner range returned %d chunks, want 1", len(innerChunks))
+	}
+	if got, want := innerChunks[0].ETag, b64md5(payload[100:200]); got != want {
+		t.Errorf("inner-range chunk ETag = %q, want copied-bytes MD5 %q", got, want)
+	}
+	wantInnerMd5 := md5.Sum(payload[100:200])
+	if !bytes.Equal(innerMd5, wantInnerMd5[:]) {
+		t.Errorf("inner-range part Md5 = %x, want %x", innerMd5, wantInnerMd5)
+	}
+}
+
+// A range over a sparse source zero-fills the gaps, so the part MD5 has to
+// hash the logical range — not just the one chunk that overlaps it.
+func TestCopyChunksForRangeSparseETags(t *testing.T) {
+	volume := startFakeVolumeServer(t)
+	filerImpl := &ambiguousPutFiler{volume: volume, entries: map[string]*filer_pb.Entry{}}
+	s3a := newPutTestServer(t, startFakeFiler(t, filerImpl))
+	util_http.InitGlobalHttpClient()
+
+	fid := fmt.Sprintf("3,%016x%08x", 105, 206)
+	volume.stored[fid] = []byte("abc")
+	sum := md5.Sum([]byte("abc"))
+	entry := &filer_pb.Entry{
+		Name: "sparse",
+		Attributes: &filer_pb.FuseAttributes{
+			FileSize: 6,
+		},
+		Chunks: []*filer_pb.FileChunk{
+			{FileId: fid, Offset: 3, Size: 3, ETag: base64.StdEncoding.EncodeToString(sum[:])},
+		},
+	}
+
+	// bytes 0-5 read as "\x00\x00\x00abc"; the part MD5 must cover the zeros too.
+	dstChunks, partMd5, err := s3a.copyChunksForRange(entry, 0, 5, "/dst")
+	if err != nil {
+		t.Fatalf("copyChunksForRange sparse: %v", err)
+	}
+	if len(dstChunks) != 1 || dstChunks[0].Offset != 3 {
+		t.Fatalf("sparse copy chunks = %+v, want one chunk at offset 3", dstChunks)
+	}
+	wantMd5 := md5.Sum([]byte("\x00\x00\x00abc"))
+	if !bytes.Equal(partMd5, wantMd5[:]) {
+		t.Errorf("sparse part Md5 = %x, want %x (zero-filled range)", partMd5, wantMd5)
+	}
+
+	// A range over a pure hole copies nothing but still reads as zeros.
+	holeChunks, holeMd5, err := s3a.copyChunksForRange(entry, 0, 2, "/dst")
+	if err != nil {
+		t.Fatalf("copyChunksForRange hole: %v", err)
+	}
+	if len(holeChunks) != 0 {
+		t.Fatalf("hole copy returned %d chunks, want 0", len(holeChunks))
+	}
+	wantHoleMd5 := md5.Sum([]byte{0, 0, 0})
+	if !bytes.Equal(holeMd5, wantHoleMd5[:]) {
+		t.Errorf("hole part Md5 = %x, want %x (all zeros)", holeMd5, wantHoleMd5)
+	}
+}
+
+// Destination chunks must resolve overlaps the way the source did: reads
+// pick winners by ModifiedTsNs, so the copied chunks keep the source's
+// precedence even though they were uploaded concurrently.
+func TestCopyChunksForRangePreservesSourcePrecedence(t *testing.T) {
+	volume := startFakeVolumeServer(t)
+	filerImpl := &ambiguousPutFiler{volume: volume, entries: map[string]*filer_pb.Entry{}}
+	s3a := newPutTestServer(t, startFakeFiler(t, filerImpl))
+	util_http.InitGlobalHttpClient()
+
+	etagOf := func(b []byte) string {
+		sum := md5.Sum(b)
+		return base64.StdEncoding.EncodeToString(sum[:])
+	}
+	fidA := fmt.Sprintf("3,%016x%08x", 107, 208)
+	fidB := fmt.Sprintf("3,%016x%08x", 109, 210)
+	volume.stored[fidA] = []byte("aaaa")
+	volume.stored[fidB] = []byte("bbbb")
+
+	// A covers [0,4), B covers [2,6); B's range is copied into a chunk that
+	// overlaps A's copy on [2,4), and the source says A wins there.
+	entry := &filer_pb.Entry{
+		Name:       "overlap",
+		Attributes: &filer_pb.FuseAttributes{FileSize: 6},
+		Chunks: []*filer_pb.FileChunk{
+			{FileId: fidA, Offset: 0, Size: 4, ModifiedTsNs: 20, ETag: etagOf([]byte("aaaa"))},
+			{FileId: fidB, Offset: 2, Size: 4, ModifiedTsNs: 10, ETag: etagOf([]byte("bbbb"))},
+		},
+	}
+	dstChunks, partMd5, err := s3a.copyChunksForRange(entry, 0, 5, "/dst")
+	if err != nil {
+		t.Fatalf("copyChunksForRange overlap: %v", err)
+	}
+	if len(dstChunks) != 2 {
+		t.Fatalf("overlap copy returned %d chunks, want 2", len(dstChunks))
+	}
+	if dstChunks[0].ModifiedTsNs <= dstChunks[1].ModifiedTsNs {
+		t.Errorf("destination precedence inverted: chunk from ts=20 got %d, chunk from ts=10 got %d",
+			dstChunks[0].ModifiedTsNs, dstChunks[1].ModifiedTsNs)
+	}
+	// The source resolves to "aaaa"+"bb"; the part MD5 must match that view.
+	wantMd5 := md5.Sum([]byte("aaaabb"))
+	if !bytes.Equal(partMd5, wantMd5[:]) {
+		t.Errorf("overlap part Md5 = %x, want %x", partMd5, wantMd5)
+	}
+}
+
+// An inline-Content source has no chunks for the raw-byte copy; the plaintext
+// stream path must write the range's real bytes and tag the part with their
+// MD5, not a zero-fill digest.
+func TestCopyObjectPartInlineContent(t *testing.T) {
+	volume := startFakeVolumeServer(t)
+	filerImpl := &ambiguousPutFiler{volume: volume, apply: true, entries: map[string]*filer_pb.Entry{}}
+	s3a := newPutTestServer(t, startFakeFiler(t, filerImpl))
+	util_http.InitGlobalHttpClient()
+
+	content := []byte("inline-content")
+	entry := &filer_pb.Entry{
+		Name:       "src",
+		Attributes: &filer_pb.FuseAttributes{FileSize: uint64(len(content))},
+		Content:    content,
+	}
+	r := httptest.NewRequest(http.MethodPut, "/b/dst?partNumber=1&uploadId=u1", nil)
+	etag, _, code := s3a.copyObjectPartViaReencryption(r, entry, 2, 5, "b", "dst", "u1", 1, &filer_pb.Entry{})
+	if code != s3err.ErrNone {
+		t.Fatalf("copyObjectPartViaReencryption: %v", code)
+	}
+	wantMd5 := md5.Sum(content[2:6])
+	if got := strings.Trim(etag, `"`); got != fmt.Sprintf("%x", wantMd5) {
+		t.Errorf("etag = %q, want %x", got, wantMd5)
+	}
+	var partEntry *filer_pb.Entry
+	for key, e := range filerImpl.entries {
+		if strings.Contains(key, "u1") {
+			partEntry = e
+		}
+	}
+	if partEntry == nil || len(partEntry.Chunks) == 0 {
+		t.Fatalf("no part entry with chunks stored: %+v", filerImpl.entries)
+	}
+	found := false
+	for _, stored := range volume.stored {
+		if bytes.Contains(stored, content[2:6]) {
+			found = true
+		}
+	}
+	if !found {
+		t.Errorf("destination volume does not hold the copied content bytes")
+	}
+}
+
+// An empty part must carry the empty-body MD5 on the stored entry too, so
+// ListParts and complete's composite ETag see the same digest CopyPartResult
+// reported.
+func TestWriteEmptyCopyPartStoresMd5(t *testing.T) {
+	volume := startFakeVolumeServer(t)
+	filerImpl := &ambiguousPutFiler{volume: volume, apply: true, entries: map[string]*filer_pb.Entry{}}
+	s3a := newPutTestServer(t, startFakeFiler(t, filerImpl))
+	util_http.InitGlobalHttpClient()
+
+	etag, code := s3a.writeEmptyCopyPart("b", "u1", 3)
+	if code != s3err.ErrNone {
+		t.Fatalf("writeEmptyCopyPart: %v", code)
+	}
+	const emptyMD5Hex = "d41d8cd98f00b204e9800998ecf8427e"
+	if etag != emptyMD5Hex {
+		t.Fatalf("writeEmptyCopyPart etag = %q, want %q", etag, emptyMD5Hex)
+	}
+	var partEntry *filer_pb.Entry
+	for key, e := range filerImpl.entries {
+		if strings.Contains(key, "u1") {
+			partEntry = e
+		}
+	}
+	if partEntry == nil {
+		t.Fatalf("no part entry stored: %+v", filerImpl.entries)
+	}
+	if got := string(partEntry.Extended[s3_constants.ExtETagKey]); got != emptyMD5Hex {
+		t.Errorf("stored ExtETagKey = %q, want %q", got, emptyMD5Hex)
+	}
+	if got := fmt.Sprintf("%x", partEntry.Attributes.Md5); got != emptyMD5Hex {
+		t.Errorf("stored Attributes.Md5 = %q, want %q", got, emptyMD5Hex)
+	}
+}
+
+// A whole-range chunk copy normally reuses the source chunk's ETag as the
+// part MD5; TUS and mount writes leave chunks without one, so the part has
+// to hash the copied bytes instead of saving an empty digest.
+func TestCopyChunksForRangeMissingChunkETag(t *testing.T) {
+	volume := startFakeVolumeServer(t)
+	filerImpl := &ambiguousPutFiler{volume: volume, entries: map[string]*filer_pb.Entry{}}
+	s3a := newPutTestServer(t, startFakeFiler(t, filerImpl))
+	util_http.InitGlobalHttpClient()
+
+	fid := fmt.Sprintf("3,%016x%08x", 111, 212)
+	volume.stored[fid] = []byte("etagless")
+	entry := &filer_pb.Entry{
+		Name:       "src",
+		Attributes: &filer_pb.FuseAttributes{FileSize: 8},
+		Chunks: []*filer_pb.FileChunk{
+			{FileId: fid, Offset: 0, Size: 8},
+		},
+	}
+
+	dstChunks, partMd5, err := s3a.copyChunksForRange(entry, 0, 7, "/dst")
+	if err != nil {
+		t.Fatalf("copyChunksForRange: %v", err)
+	}
+	if len(dstChunks) != 1 {
+		t.Fatalf("copyChunksForRange returned %d chunks, want 1", len(dstChunks))
+	}
+	wantMd5 := md5.Sum([]byte("etagless"))
+	if !bytes.Equal(partMd5, wantMd5[:]) {
+		t.Errorf("part Md5 = %x, want %x (hashed copied range)", partMd5, wantMd5)
+	}
+}

weed/s3api/s3api_object_handlers_copy_part_sse.go

@@ -2,6 +2,7 @@ package s3api
 
 import (
 	"context"
+	"crypto/md5"
 	"encoding/base64"
 	"encoding/json"
 	"errors"
@@ -419,14 +420,19 @@ func (s3a *S3ApiServer) writeEmptyCopyPart(dstBucket, uploadID string, partID in
 			return "", s3err.ErrInternalError
 		}
 	}
+	const emptyMD5Hex = "d41d8cd98f00b204e9800998ecf8427e"
 	if err := s3a.mkFile(uploadDir, partName, nil, func(e *filer_pb.Entry) {
 		if e.Attributes == nil {
 			e.Attributes = &filer_pb.FuseAttributes{}
 		}
 		e.Attributes.FileSize = 0
+		e.Attributes.Md5 = md5.New().Sum(nil)
+		if e.Extended == nil {
+			e.Extended = map[string][]byte{}
+		}
+		e.Extended[s3_constants.ExtETagKey] = []byte(emptyMD5Hex)
 	}); err != nil {
 		return "", s3err.ErrInternalError
 	}
-	const emptyMD5Hex = "d41d8cd98f00b204e9800998ecf8427e"
 	return emptyMD5Hex, s3err.ErrNone
 }

weed/s3api/s3api_object_handlers_copy_stream.go

@@ -73,12 +73,19 @@ func canStreamCopyChunk(chunk *filer_pb.FileChunk) bool {
 // Trusting a hint that disagrees with the wire bytes (e.g. caller says
 // "compressed" but the volume returned raw bytes for a partial range)
 // would label raw bytes as gzip and corrupt reads on the destination.
+//
+// digestWriter, when non-nil, receives the plaintext bytes as they are
+// forwarded so the caller can derive the destination chunk's ETag. It is
+// not fed when the wire bytes are gzip — only possible on the full-chunk
+// pass-through, whose logical content still matches the source chunk's
+// own ETag and needs no recomputation.
 func (s3a *S3ApiServer) streamCopyChunkRange(
 	ctx context.Context,
 	srcUrl, srcFileId string,
 	offset, size int64,
 	isFullChunk bool,
 	assignResult *filer_pb.AssignVolumeResponse,
+	digestWriter io.Writer,
 ) error {
 	// Range header takes int64, but mirror the int32 bound from
 	// downloadChunkData so the two paths reject the same inputs.
@@ -166,7 +173,11 @@ func (s3a *S3ApiServer) streamCopyChunkRange(
 			return
 		}
 
-		if _, err := io.Copy(fw, srcResp.Body); err != nil {
+		src := io.Reader(srcResp.Body)
+		if digestWriter != nil && !bodyIsGzipped {
+			src = io.TeeReader(srcResp.Body, digestWriter)
+		}
+		if _, err := io.Copy(fw, src); err != nil {
 			if shouldLogStreamError(err) {
 				glog.V(2).Infof("stream copy %s offset=%d size=%d: %v",
 					srcUrl, offset, size, err)

weed/s3api/s3api_object_handlers_readonly_test.go

@@ -87,7 +87,7 @@ func TestCopyChunksReadOnlyIsAccessDenied(t *testing.T) {
 		t.Fatalf("copyChunks err %v maps to %v, want %v", err, code, s3err.ErrAccessDenied)
 	}
 
-	_, err = s3a.copyChunksForRange(entry, 0, 7, "/buckets/b/dst")
+	_, _, err = s3a.copyChunksForRange(entry, 0, 7, "/buckets/b/dst")
 	if code := s3a.mapCopyErrorToS3Error(err); code != s3err.ErrAccessDenied {
 		t.Fatalf("copyChunksForRange err %v maps to %v, want %v", err, code, s3err.ErrAccessDenied)
 	}

weed/s3api/s3api_object_range_test.go

@@ -40,3 +40,33 @@ func TestParseAndValidateRangeRejectsMalformedNumericOffsets(t *testing.T) {
 		})
 	}
 }
+
+func TestHeadObjectRange(t *testing.T) {
+	s3a, f := newAnonymousReadTestServer(t)
+	f.entries["/buckets/b/o"] = anonymousReadEntry(anonymousReadGrant("READ"))
+	for _, tc := range []struct {
+		rangeHeader, contentRange, contentLength string
+		code                                     int
+	}{
+		{"", "", "11", http.StatusOK},
+		{"bytes=1-4", "bytes 1-4/11", "4", http.StatusPartialContent},
+		{"bytes=-3", "bytes 8-10/11", "3", http.StatusPartialContent},
+		{"bytes=5-", "bytes 5-10/11", "6", http.StatusPartialContent},
+		{"bytes=20-", "bytes */11", "", http.StatusRequestedRangeNotSatisfiable},
+	} {
+		headers := map[string]string{}
+		if tc.rangeHeader != "" {
+			headers["Range"] = tc.rangeHeader
+		}
+		rr := serveAnonymousRead(s3a, http.MethodHead, "o", "", headers)
+		if rr.Code != tc.code {
+			t.Errorf("Range %q: status %d, want %d", tc.rangeHeader, rr.Code, tc.code)
+		}
+		if got := rr.Header().Get("Content-Range"); got != tc.contentRange {
+			t.Errorf("Range %q: Content-Range %q, want %q", tc.rangeHeader, got, tc.contentRange)
+		}
+		if got := rr.Header().Get("Content-Length"); tc.contentLength != "" && got != tc.contentLength {
+			t.Errorf("Range %q: Content-Length %q, want %q", tc.rangeHeader, got, tc.contentLength)
+		}
+	}
+}

weed/s3api/s3tables/handler.go

@@ -459,6 +459,70 @@ func isAuthError(err error) bool {
 	return errors.As(err, &authErr) || errors.Is(err, ErrAccessDenied)
 }
 
+// authorizeAbsentTarget reports whether the caller may learn the target
+// resolved to nothing, so a caller without a grant gets the same response a
+// denied existing target would give. The bucket named by the target still
+// applies whatever metadata and policy load; with nothing readable to
+// evaluate, only an identity-level grant on the operation, the admin, or a
+// default-allow deployment passes.
+func (h *S3TablesHandler) authorizeAbsentTarget(r *http.Request, filerClient FilerClient, operation, bucketName, namespace, name, resourceARN string) bool {
+	principal := h.getAccountID(r)
+	if principal == "" {
+		return false
+	}
+
+	var bucketOwner, bucketPolicy string
+	var bucketTags map[string]string
+	if bucketName != "" {
+		bucketPath := GetTableBucketPath(bucketName)
+		_ = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+			if data, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata); err == nil {
+				var meta tableBucketMetadata
+				if json.Unmarshal(data, &meta) == nil {
+					bucketOwner = meta.OwnerAccountID
+				}
+			}
+			if policy, err := h.readPolicy(r, client, bucketPath); err == nil {
+				bucketPolicy = policy
+			}
+			bucketTags, _ = h.readTags(r.Context(), client, bucketPath)
+			return nil
+		})
+	}
+
+	bucketARN := h.generateTableBucketARN(bucketOwner, bucketName)
+	// Judge the target by the canonical ARN the surviving bucket resolves to:
+	// the request ARN's account and region are the caller's to choose and must
+	// not decide whether a policy Deny applies.
+	canonicalARN := resourceARN
+	if bucketOwner != "" {
+		if name != "" {
+			canonicalARN = h.generateTableARN(bucketOwner, bucketName, namespace+"/"+name)
+		} else {
+			canonicalARN = bucketARN
+		}
+	}
+	ctx := &PolicyContext{
+		TrustedIssuers:   h.trustedIssuers(r),
+		IsLocalAccountID: h.localAccountIDs,
+		TableBucketName:  bucketName,
+		Namespace:        namespace,
+		TableName:        name,
+		TableBucketTags:  bucketTags,
+		IdentityActions:  getIdentityActions(r),
+		DefaultAllow:     h.defaultAllowFor(r),
+	}
+	if resourceARN != "" &&
+		matchesDenyStatement(operation, principal, bucketPolicy, canonicalARN, ctx) {
+		return false
+	}
+	if CheckTableOrBucketPermission(operation, principal, "", "", canonicalARN,
+		bucketOwner, bucketPolicy, bucketARN, ctx) {
+		return true
+	}
+	return bucketOwner == "" && checkPermission(operation, principal, "", "", resourceARN, ctx)
+}
+
 func (h *S3TablesHandler) readTags(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) (map[string]string, error) {
 	data, err := h.getExtendedAttribute(ctx, client, path, ExtendedKeyTags)
 	if err != nil {

weed/s3api/s3tables/handler_bucket_get_list_delete.go

@@ -58,7 +58,7 @@ func (h *S3TablesHandler) handleGetTableBucket(w http.ResponseWriter, r *http.Re
 	})
 
 	if err != nil {
-		if errors.Is(err, filer_pb.ErrNotFound) {
+		if errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, ErrAttributeNotFound) {
 			h.writeError(w, http.StatusNotFound, ErrCodeNoSuchBucket, fmt.Sprintf("table bucket %s not found", bucketName))
 		} else {
 			h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to get table bucket: %v", err))

weed/s3api/s3tables/handler_existence_hiding_test.go

@@ -3,6 +3,7 @@ package s3tables
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"net/http/httptest"
 	"testing"
 
@@ -211,3 +212,165 @@ func TestDeniedTableWritesDoNotRevealViews(t *testing.T) {
 		})
 	}
 }
+
+// A caller without a grant must get the same answer whether a list target
+// exists or not, and whether the bucket behind it exists at all.
+func TestAbsentListTargetsMatchDeniedAnswers(t *testing.T) {
+	existing, manager := startRenameManager(t)
+	missing := s3tablestest.Start(t)
+	manager.SetTrusted(false)
+	manager.SetDefaultAllow(false)
+
+	for _, tc := range []struct {
+		name      string
+		fs        *s3tablestest.MemFiler
+		operation string
+		denied    interface{}
+		request   interface{}
+	}{
+		{"ListTables missing namespace", existing, "ListTables",
+			&ListTablesRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}},
+			&ListTablesRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ghost"}}},
+		{"ListTables missing bucket", missing, "ListTables",
+			&ListTablesRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}},
+			&ListTablesRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}}},
+		{"ListTables missing bucket, all namespaces", missing, "ListTables",
+			&ListTablesRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}},
+			&ListTablesRequest{TableBucketARN: mustBucketARN(t)}},
+		{"ListViews missing namespace", existing, "ListViews",
+			&ListViewsRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}},
+			&ListViewsRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ghost"}}},
+		{"CreateNamespace missing bucket", missing, "CreateNamespace",
+			&CreateNamespaceRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"newns"}},
+			&CreateNamespaceRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"newns"}}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			denied := runUnauthorizedRequest(t, manager, existing, tc.operation, tc.denied)
+			assert.Equal(t, 403, denied.status)
+			got := runUnauthorizedRequest(t, manager, tc.fs, tc.operation, tc.request)
+			assert.Equal(t, denied, got)
+		})
+	}
+}
+
+// A denied tag operation and the same operation on a missing table must answer
+// alike; a granted caller still gets the truthful not-found.
+func TestAbsentTagTargetsMatchDeniedAnswers(t *testing.T) {
+	existing, manager := startRenameManager(t)
+	manager.SetTrusted(false)
+	manager.SetDefaultAllow(false)
+
+	existingARN, err := BuildTableARN(DefaultRegion, DefaultAccountID, renameTestBucket, "ns", "t")
+	require.NoError(t, err)
+	missingARN, err := BuildTableARN(DefaultRegion, DefaultAccountID, renameTestBucket, "ns", "ghost")
+	require.NoError(t, err)
+
+	requestFor := func(operation, arn string) interface{} {
+		switch operation {
+		case "TagResource":
+			return &TagResourceRequest{ResourceARN: arn, Tags: map[string]string{"k": "v"}}
+		case "UntagResource":
+			return &UntagResourceRequest{ResourceARN: arn, TagKeys: []string{"k"}}
+		}
+		return &ListTagsForResourceRequest{ResourceARN: arn}
+	}
+
+	for _, operation := range []string{"TagResource", "ListTagsForResource", "UntagResource"} {
+		t.Run(operation, func(t *testing.T) {
+			denied := runUnauthorizedRequest(t, manager, existing, operation, requestFor(operation, existingARN))
+			assert.Equal(t, 403, denied.status)
+			got := runUnauthorizedRequest(t, manager, existing, operation, requestFor(operation, missingARN))
+			assert.Equal(t, denied, got)
+		})
+	}
+}
+
+// A directory that holds a plain S3 bucket name is not a table bucket: every
+// caller gets the same 404 an absent name gets, not a 500 that reveals the
+// name resolves in the filer.
+func TestPlainBucketReadsAsAbsent(t *testing.T) {
+	existing, manager := startRenameManager(t)
+	manager.SetTrusted(false)
+	manager.SetDefaultAllow(false)
+	existing.Put(TablesPath, "plainbkt", nil)
+
+	plainARN, err := BuildBucketARN(DefaultRegion, DefaultAccountID, "plainbkt")
+	require.NoError(t, err)
+
+	for _, tc := range []struct {
+		operation string
+		request   interface{}
+	}{
+		{"GetTableBucket", &GetTableBucketRequest{TableBucketARN: plainARN}},
+		{"ListNamespaces", &ListNamespacesRequest{TableBucketARN: plainARN}},
+	} {
+		t.Run(tc.operation, func(t *testing.T) {
+			got := runUnauthorizedRequest(t, manager, existing, tc.operation, tc.request)
+			assert.Equal(t, 404, got.status)
+			assert.Equal(t, ErrCodeNoSuchBucket, got.body.Type)
+		})
+	}
+}
+
+// A caller with an identity-level grant still learns that a name is absent.
+func TestGrantedCallerLearnsAbsence(t *testing.T) {
+	missing := s3tablestest.Start(t)
+	manager := NewManager()
+	manager.SetTrusted(false)
+	manager.SetDefaultAllow(false)
+
+	got := runRequestAs(t, manager, missing, "ListTables", &ListTablesRequest{
+		TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}},
+		[]string{"s3tables:ListTables"})
+	assert.Equal(t, 200, got.status)
+
+	missingTableARN, err := BuildTableARN(DefaultRegion, DefaultAccountID, renameTestBucket, "ns", "ghost")
+	require.NoError(t, err)
+	got = runRequestAs(t, manager, missing, "ListTagsForResource",
+		&ListTagsForResourceRequest{ResourceARN: missingTableARN},
+		[]string{"s3tables:ListTagsForResource"})
+	assert.Equal(t, 404, got.status)
+	assert.Equal(t, ErrCodeNoSuchTable, got.body.Type)
+}
+
+// A request ARN that names a foreign account still resolves to the local
+// names, so the absent-target answer must match the denial the canonical ARN
+// gets; otherwise a reformatted ARN restores the existence oracle.
+func TestAbsentTagTargetDenySurvivesForeignAccountARN(t *testing.T) {
+	fs, manager := startRenameManager(t)
+	manager.SetTrusted(false)
+	manager.SetDefaultAllow(false)
+
+	bucketMeta, _ := json.Marshal(tableBucketMetadata{Name: renameTestBucket, OwnerAccountID: DefaultAccountID})
+	fs.Put(TablesPath, renameTestBucket, map[string][]byte{
+		ExtendedKeyTableBucket: []byte("{}"),
+		ExtendedKeyMetadata:    bucketMeta,
+		ExtendedKeyPolicy: []byte(fmt.Sprintf(
+			`{"Statement":[{"Effect":"Deny","Principal":"*","Action":"s3tables:*","Resource":%q}]}`,
+			mustBucketARN(t)+"/table/*")),
+	})
+
+	foreignARN := func(table string) string {
+		return fmt.Sprintf("arn:aws:s3tables:%s:999999999999:bucket/%s/table/ns/%s",
+			DefaultRegion, renameTestBucket, table)
+	}
+	requestFor := func(operation, arn string) interface{} {
+		switch operation {
+		case "TagResource":
+			return &TagResourceRequest{ResourceARN: arn, Tags: map[string]string{"k": "v"}}
+		case "UntagResource":
+			return &UntagResourceRequest{ResourceARN: arn, TagKeys: []string{"k"}}
+		}
+		return &ListTagsForResourceRequest{ResourceARN: arn}
+	}
+
+	for _, operation := range []string{"TagResource", "ListTagsForResource", "UntagResource"} {
+		t.Run(operation, func(t *testing.T) {
+			actions := []string{"s3tables:" + operation}
+			denied := runRequestAs(t, manager, fs, operation, requestFor(operation, foreignARN("t")), actions)
+			assert.Equal(t, 403, denied.status)
+			got := runRequestAs(t, manager, fs, operation, requestFor(operation, foreignARN("ghost")), actions)
+			assert.Equal(t, denied, got)
+		})
+	}
+}

weed/s3api/s3tables/handler_namespace.go

@@ -140,6 +140,10 @@ func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.R
 			return ErrAccessDenied
 		}
 		if errors.Is(err, filer_pb.ErrNotFound) {
+			if !h.authorizeAbsentTarget(r, filerClient, "CreateNamespace", bucketName, namespaceName, "", "") {
+				h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to create namespace in this bucket")
+				return ErrAccessDenied
+			}
 			h.writeError(w, http.StatusNotFound, ErrCodeNoSuchBucket, fmt.Sprintf("table bucket %s not found", bucketName))
 		} else {
 			h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to check table bucket: %v", err))
@@ -481,7 +485,7 @@ func (h *S3TablesHandler) handleListNamespaces(w http.ResponseWriter, r *http.Re
 	})
 
 	if err != nil {
-		if errors.Is(err, filer_pb.ErrNotFound) {
+		if errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, ErrAttributeNotFound) {
 			h.writeError(w, http.StatusNotFound, ErrCodeNoSuchBucket, fmt.Sprintf("table bucket %s not found", bucketName))
 		} else {
 			h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to list namespaces: %v", err))

weed/s3api/s3tables/handler_policy.go

@@ -659,6 +659,9 @@ func (h *S3TablesHandler) handleTagResource(w http.ResponseWriter, r *http.Reque
 		// Read metadata for ownership check
 		data, err := h.getExtendedAttribute(r.Context(), client, resourcePath, ExtendedKeyMetadata)
 		if err != nil {
+			if errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, ErrAttributeNotFound) {
+				return h.absentTagTargetError(r, client, "TagResource", "not authorized to tag resource", req.ResourceARN, rType)
+			}
 			return err
 		}
 
@@ -793,6 +796,9 @@ func (h *S3TablesHandler) handleListTagsForResource(w http.ResponseWriter, r *ht
 		// Read metadata for ownership check
 		data, err := h.getExtendedAttribute(r.Context(), client, resourcePath, ExtendedKeyMetadata)
 		if err != nil {
+			if errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, ErrAttributeNotFound) {
+				return h.absentTagTargetError(r, client, "ListTagsForResource", "not authorized to list tags for resource", req.ResourceARN, rType)
+			}
 			return err
 		}
 
@@ -913,6 +919,9 @@ func (h *S3TablesHandler) handleUntagResource(w http.ResponseWriter, r *http.Req
 		// Read metadata for ownership check
 		data, err := h.getExtendedAttribute(r.Context(), client, resourcePath, ExtendedKeyMetadata)
 		if err != nil {
+			if errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, ErrAttributeNotFound) {
+				return h.absentTagTargetError(r, client, "UntagResource", "not authorized to untag resource", req.ResourceARN, rType)
+			}
 			return err
 		}
 
@@ -1017,6 +1026,22 @@ func (h *S3TablesHandler) handleUntagResource(w http.ResponseWriter, r *http.Req
 	return nil
 }
 
+// absentTagTargetError resolves a tag operation when the resource's catalog
+// metadata did not load: a caller the operation would refuse gets its denial,
+// everyone else the same not-found an absent name gets.
+func (h *S3TablesHandler) absentTagTargetError(r *http.Request, client filer_pb.SeaweedFilerClient, operation, deniedMessage, resourceARN string, rType ResourceType) error {
+	var bucketName, namespace, name string
+	if rType == ResourceTypeTable {
+		bucketName, namespace, name, _ = parseTableFromARN(resourceARN)
+	} else {
+		bucketName, _ = parseBucketNameFromARN(resourceARN)
+	}
+	if !h.authorizeAbsentTarget(r, &singleClient{client}, operation, bucketName, namespace, name, resourceARN) {
+		return NewAuthError(operation, h.getAccountID(r), deniedMessage)
+	}
+	return filer_pb.ErrNotFound
+}
+
 // resolveResourcePath determines the resource path and extended attribute key from a resource ARN
 func (h *S3TablesHandler) resolveResourcePath(resourceARN string) (path string, key string, rType ResourceType, err error) {
 	// Try parsing as table ARN first

weed/s3api/s3tables/handler_table.go

@@ -515,11 +515,14 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 			bucketTags := map[string]string{}
 
 			// Fetch namespace metadata and policy
+			targetAbsent := false
 			data, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyMetadata)
 			if err != nil {
-				return err // Not Found handled by caller
-			}
-			if err := json.Unmarshal(data, &nsMeta); err != nil {
+				if !errors.Is(err, filer_pb.ErrNotFound) && !errors.Is(err, ErrAttributeNotFound) {
+					return err
+				}
+				targetAbsent = true
+			} else if err := json.Unmarshal(data, &nsMeta); err != nil {
 				return err
 			}
 
@@ -527,7 +530,7 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 			policyData, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyPolicy)
 			if err == nil {
 				namespacePolicy = string(policyData)
-			} else if !errors.Is(err, ErrAttributeNotFound) {
+			} else if !errors.Is(err, ErrAttributeNotFound) && !errors.Is(err, filer_pb.ErrNotFound) {
 				return fmt.Errorf("failed to fetch namespace policy: %w", err)
 			}
 
@@ -537,6 +540,8 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 				if err := json.Unmarshal(data, &bucketMeta); err != nil {
 					return fmt.Errorf("failed to unmarshal bucket metadata: %w", err)
 				}
+			} else if errors.Is(err, filer_pb.ErrNotFound) {
+				targetAbsent = true
 			} else if !errors.Is(err, ErrAttributeNotFound) {
 				return fmt.Errorf("failed to fetch bucket metadata: %w", err)
 			}
@@ -544,15 +549,24 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 			policyData, err = h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
 			if err == nil {
 				bucketPolicy = string(policyData)
-			} else if !errors.Is(err, ErrAttributeNotFound) {
+			} else if !errors.Is(err, ErrAttributeNotFound) && !errors.Is(err, filer_pb.ErrNotFound) {
 				return fmt.Errorf("failed to fetch bucket policy: %w", err)
 			}
 			if tags, err := h.readTags(r.Context(), client, bucketPath); err != nil {
-				return fmt.Errorf("failed to read bucket tags: %w", err)
+				if !errors.Is(err, filer_pb.ErrNotFound) {
+					return fmt.Errorf("failed to read bucket tags: %w", err)
+				}
 			} else if tags != nil {
 				bucketTags = tags
 			}
 
+			if targetAbsent {
+				if !h.authorizeAbsentTarget(r, &singleClient{client}, "ListTables", bucketName, namespaceName, "", "") {
+					return ErrAccessDenied
+				}
+				return filer_pb.ErrNotFound
+			}
+
 			bucketARN := h.generateTableBucketARN(bucketMeta.OwnerAccountID, bucketName)
 			identityActions := getIdentityActions(r)
 			if !CheckTableOrBucketPermission("ListTables", accountID,
@@ -583,7 +597,13 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 			// Fetch bucket metadata and policy
 			data, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata)
 			if err != nil {
-				return fmt.Errorf("failed to fetch bucket metadata: %w", err)
+				if !errors.Is(err, filer_pb.ErrNotFound) && !errors.Is(err, ErrAttributeNotFound) {
+					return fmt.Errorf("failed to fetch bucket metadata: %w", err)
+				}
+				if !h.authorizeAbsentTarget(r, &singleClient{client}, "ListTables", bucketName, "", "", "") {
+					return ErrAccessDenied
+				}
+				return filer_pb.ErrNotFound
 			}
 			if err := json.Unmarshal(data, &bucketMeta); err != nil {
 				return fmt.Errorf("failed to unmarshal bucket metadata: %w", err)

weed/s3api/s3tables/handler_view.go

@@ -279,7 +279,13 @@ func (h *S3TablesHandler) handleListViews(w http.ResponseWriter, r *http.Request
 		var nsMeta namespaceMetadata
 		data, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyMetadata)
 		if err != nil {
-			return err
+			if !errors.Is(err, filer_pb.ErrNotFound) && !errors.Is(err, ErrAttributeNotFound) {
+				return err
+			}
+			if !h.authorizeAbsentTarget(r, &singleClient{client}, "ListViews", bucketName, namespaceName, "", "") {
+				return ErrAccessDenied
+			}
+			return filer_pb.ErrNotFound
 		}
 		if err := json.Unmarshal(data, &nsMeta); err != nil {
 			return err

weed/server/volume_grpc_client_to_master.go

@@ -54,6 +54,22 @@ func (vs *VolumeServer) lookupRaftLeaderMaster(ctx context.Context) (pb.ServerAd
 	return leader, nil
 }
 
+// refreshLeaderFromPeers publishes the leader the configured masters report,
+// so lookups stop going to a master the heartbeat just lost while the loop
+// retries. Without an answer within timeout no leader is published: the lost
+// master is dead or no longer leader, and a follower answers lookups from its
+// own cache rather than refusing them, so failing fast is the safer answer.
+func (vs *VolumeServer) refreshLeaderFromPeers(timeout time.Duration) {
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	leader, err := vs.lookupRaftLeaderMaster(ctx)
+	if err != nil {
+		vs.setCurrentMaster("")
+		return
+	}
+	vs.setCurrentMaster(leader)
+}
+
 func isContextDoneErr(err error) bool {
 	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
 		return true
@@ -74,7 +90,10 @@ func (vs *VolumeServer) getCurrentMaster() pb.ServerAddress {
 }
 
 // setCurrentMaster updates vs.currentMaster under a write lock. The
-// heartbeat goroutine calls this whenever it (re)connects to a master.
+// heartbeat goroutine calls this only with a master known to be the leader:
+// one that answered a heartbeat as leader, or the one a follower named, on
+// the heartbeat stream or, after the stream failed, in its configuration.
+// After a failed stream with no leader named, it is cleared to "".
 func (vs *VolumeServer) setCurrentMaster(master pb.ServerAddress) {
 	vs.currentMasterLock.Lock()
 	vs.currentMaster = master
@@ -122,7 +141,6 @@ func (vs *VolumeServer) heartbeat() {
 				time.Sleep(3 * time.Second)
 				master = newLeader
 			}
-			vs.store.MasterAddress = master
 			newLeader, err = vs.doHeartbeatWithRetry(master, grpcDialOption, vs.pulsePeriod, duplicateRetryCount)
 			if err != nil {
 				glog.V(0).Infof("heartbeat to %s error: %v", master, err)
@@ -136,12 +154,15 @@ func (vs *VolumeServer) heartbeat() {
 				} else {
 					// Regular error, reset duplicate retry count
 					duplicateRetryCount = 0
-					time.Sleep(vs.pulsePeriod)
+					// The lookup spends part of the retry delay rather than
+					// adding to it.
+					retryAt := time.Now().Add(vs.pulsePeriod)
+					vs.refreshLeaderFromPeers(vs.pulsePeriod)
+					time.Sleep(time.Until(retryAt))
 				}
 
 				stats.VolumeServerMasterDisconnections.WithLabelValues(master.String()).Inc()
 				newLeader = ""
-				vs.store.MasterAddress = ""
 			} else {
 				// Successful connection, reset retry count
 				duplicateRetryCount = 0
@@ -180,17 +201,42 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp
 		return "", err
 	}
 	glog.V(0).Infof("Heartbeat to: %v", masterAddress)
-	vs.setCurrentMaster(masterAddress)
 
 	doneChan := make(chan error, 1)
+	// The receiver asks for a fresh heartbeat here rather than sending it,
+	// since a gRPC stream allows only one sender at a time.
+	resendChan := make(chan struct{}, 1)
 
 	go func() {
+		confirmedLeader := false
 		for {
 			in, err := stream.Recv()
 			if err != nil {
 				doneChan <- err
 				return
 			}
+			// A reply names the leader or comes from it, so it settles where
+			// lookups go before anything else on the reply is handled: a
+			// follower reply carries no other fields, and a leader stays the
+			// lookup target even when it rejects this registration.
+			if in.GetLeader() != "" {
+				// Only a follower names the leader. Publish the named leader right
+				// away, so lookups stop landing on this follower while the loop
+				// reconnects.
+				leader := pb.ServerAddress(in.GetLeader())
+				if !masterAddress.Equals(leader) {
+					glog.V(0).Infof("Volume Server found a new master newLeader: %v instead of %v", leader, masterAddress)
+					vs.setCurrentMaster(leader)
+					newLeader = leader
+					doneChan <- nil
+					return
+				}
+			} else if !confirmedLeader {
+				// The leader answers every new heartbeat stream without naming a
+				// leader, so this peer is the one lookups should go to.
+				confirmedLeader = true
+				vs.setCurrentMaster(masterAddress)
+			}
 			if len(in.DuplicatedUuids) > 0 {
 				var duplicateDir []string
 				for _, loc := range vs.store.Locations {
@@ -231,9 +277,9 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp
 			}
 			if volumeOptsChanged {
 				if vs.store.MaybeAdjustVolumeMax() {
-					if err = stream.Send(vs.store.CollectHeartbeat()); err != nil {
-						glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", vs.getCurrentMaster(), err)
-						return
+					select {
+					case resendChan <- struct{}{}:
+					default:
 					}
 				}
 			}
@@ -244,15 +290,6 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp
 				glog.V(0).Infof("master %s asked for the full volume list", masterAddress)
 				vs.store.RequestFullVolumeList()
 			}
-			if in.GetLeader() != "" {
-				current := vs.getCurrentMaster()
-				if !current.Equals(pb.ServerAddress(in.GetLeader())) {
-					glog.V(0).Infof("Volume Server found a new master newLeader: %v instead of %v", in.GetLeader(), current)
-					newLeader = pb.ServerAddress(in.GetLeader())
-					doneChan <- nil
-					return
-				}
-			}
 		}
 	}()
 
@@ -359,6 +396,11 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp
 				glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
 				return "", err
 			}
+		case <-resendChan:
+			if err = stream.Send(vs.store.CollectHeartbeat()); err != nil {
+				glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterAddress, err)
+				return "", err
+			}
 		case <-volumeTickChan.C:
 			glog.V(4).Infof("volume server %s:%d heartbeat", vs.store.Ip, vs.store.Port)
 			vs.store.MaybeAdjustVolumeMax()

weed/server/volume_grpc_client_to_master_test.go

@@ -4,11 +4,17 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"strings"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/storage"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 )
@@ -60,3 +66,279 @@ func startFakeMasterServerForLeaderLookup(t *testing.T, srv master_pb.SeaweedSer
 	_, port, _ := net.SplitHostPort(lis.Addr().String())
 	return pb.ServerAddress(fmt.Sprintf("127.0.0.1:0.%s", port))
 }
+
+// fakeHeartbeatMaster answers SendHeartbeat the way a master does: a follower
+// names the leader on every heartbeat, the leader answers the first one with
+// its volume options and never names itself.
+type fakeHeartbeatMaster struct {
+	master_pb.UnimplementedSeaweedServer
+	leader     string        // set on a follower
+	release    chan struct{} // a follower waits on it before answering
+	heartbeats chan *master_pb.Heartbeat
+}
+
+func (m *fakeHeartbeatMaster) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServer) error {
+	answered := false
+	for {
+		hb, err := stream.Recv()
+		if err != nil {
+			return err
+		}
+		if m.heartbeats != nil {
+			m.heartbeats <- hb
+		}
+		if m.leader != "" {
+			if m.release != nil {
+				<-m.release
+			}
+			if err := stream.Send(&master_pb.HeartbeatResponse{Leader: m.leader}); err != nil {
+				return err
+			}
+			continue
+		}
+		if !answered {
+			answered = true
+			if err := stream.Send(&master_pb.HeartbeatResponse{VolumeSizeLimit: 1 << 30}); err != nil {
+				return err
+			}
+		}
+	}
+}
+
+func newHeartbeatTestVolumeServer(t *testing.T) *VolumeServer {
+	t.Helper()
+	store := storage.NewStore(nil, "127.0.0.1", 8080, 18080, "http://127.0.0.1:8080", "store-id",
+		[]string{t.TempDir()},
+		[]int32{0},
+		[]util.MinFreeSpace{{}},
+		"",
+		storage.NeedleMapInMemory,
+		[]types.DiskType{types.HardDriveType},
+		nil,
+		3,
+		stats.DefaultDiskIOProbeConfig(),
+	)
+	t.Cleanup(store.Close)
+	vs := &VolumeServer{
+		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
+		store:          store,
+		stopChan:       make(chan bool),
+		isHeartbeating: true,
+	}
+	store.MasterFn = vs.getCurrentMaster
+	return vs
+}
+
+func waitForCurrentMaster(t *testing.T, vs *VolumeServer, want pb.ServerAddress) {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for !vs.getCurrentMaster().Equals(want) {
+		if time.Now().After(deadline) {
+			t.Fatalf("currentMaster = %q, want %q", vs.getCurrentMaster(), want)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+}
+
+// heartbeatRun is one doHeartbeatWithRetry call running in the background.
+type heartbeatRun struct {
+	done   chan heartbeatResult // receives the call's result once
+	exited chan struct{}
+	stop   func()
+}
+
+type heartbeatResult struct {
+	newLeader pb.ServerAddress
+	err       error
+}
+
+// startHeartbeat runs doHeartbeatWithRetry against master until it returns or
+// is stopped. Its cleanup stops the heartbeat and waits for it, and runs before
+// the cleanups registered earlier, so a failed assertion cannot leave the stream
+// open for the fake master's GracefulStop to wait on.
+func startHeartbeat(t *testing.T, vs *VolumeServer, master pb.ServerAddress) *heartbeatRun {
+	t.Helper()
+	var stopOnce sync.Once
+	run := &heartbeatRun{
+		done:   make(chan heartbeatResult, 1),
+		exited: make(chan struct{}),
+		stop:   func() { stopOnce.Do(func() { close(vs.stopChan) }) },
+	}
+	go func() {
+		defer close(run.exited)
+		newLeader, err := vs.doHeartbeatWithRetry(master, vs.grpcDialOption, time.Hour, 0)
+		run.done <- heartbeatResult{newLeader, err}
+	}()
+	t.Cleanup(func() {
+		run.stop()
+		select {
+		case <-run.exited:
+		case <-time.After(5 * time.Second):
+			t.Error("heartbeat did not stop")
+		}
+	})
+	return run
+}
+
+// A follower must never become the master lookups go to: until it names the
+// leader the previous one stays, and then the named leader takes over before
+// the heartbeat loop has reconnected.
+func TestHeartbeatToFollowerPublishesNamedLeaderOnly(t *testing.T) {
+	const leader = pb.ServerAddress("leader.example.com:9333.19333")
+	follower := &fakeHeartbeatMaster{
+		leader:     string(leader),
+		release:    make(chan struct{}),
+		heartbeats: make(chan *master_pb.Heartbeat, 8),
+	}
+	followerAddr := startFakeMasterServerForLeaderLookup(t, follower)
+	// cleanups run last-in first-out, so a failed assertion still lets the
+	// follower answer before its server stops gracefully
+	var releaseOnce sync.Once
+	release := func() { releaseOnce.Do(func() { close(follower.release) }) }
+	t.Cleanup(release)
+
+	vs := newHeartbeatTestVolumeServer(t)
+	previous := pb.ServerAddress("previous.example.com:9333.19333")
+	vs.setCurrentMaster(previous)
+	run := startHeartbeat(t, vs, followerAddr)
+
+	select {
+	case <-follower.heartbeats:
+	case <-time.After(5 * time.Second):
+		t.Fatal("follower got no heartbeat")
+	}
+	if got := vs.getCurrentMaster(); got != previous {
+		t.Fatalf("currentMaster = %q while connected to follower %q, want unchanged %q", string(got), string(followerAddr), string(previous))
+	}
+	if got := vs.store.MasterFn(); got != previous {
+		t.Fatalf("store master = %q, want %q", got, previous)
+	}
+	release()
+
+	select {
+	case r := <-run.done:
+		if r.err != nil {
+			t.Fatalf("heartbeat: %v", r.err)
+		}
+		if r.newLeader != leader {
+			t.Fatalf("newLeader = %q, want %q", r.newLeader, leader)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("heartbeat did not return after the follower named the leader")
+	}
+	if got := vs.getCurrentMaster(); got != leader {
+		t.Fatalf("currentMaster = %q, want named leader %q", got, leader)
+	}
+}
+
+// The leader's first answer confirms it, and the size limit that answer carries
+// makes the receiver ask for a fresh heartbeat, which the send loop delivers.
+func TestHeartbeatToLeaderPublishesItAndResendsFromSendLoop(t *testing.T) {
+	leader := &fakeHeartbeatMaster{heartbeats: make(chan *master_pb.Heartbeat, 8)}
+	leaderAddr := startFakeMasterServerForLeaderLookup(t, leader)
+
+	vs := newHeartbeatTestVolumeServer(t)
+	run := startHeartbeat(t, vs, leaderAddr)
+
+	waitForCurrentMaster(t, vs, leaderAddr)
+	if got := vs.store.MasterFn(); got != leaderAddr {
+		t.Fatalf("store master = %q, want %q", got, leaderAddr)
+	}
+
+	// the initial volume and ec heartbeats, then the resend
+	for i := 0; i < 3; i++ {
+		select {
+		case <-leader.heartbeats:
+		case <-time.After(5 * time.Second):
+			t.Fatalf("leader got %d heartbeats, want 3", i)
+		}
+	}
+
+	run.stop()
+	select {
+	case r := <-run.done:
+		if r.err != nil {
+			t.Fatalf("heartbeat: %v", r.err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("heartbeat did not stop")
+	}
+}
+
+// duplicateUUIDMaster answers every heartbeat the way a leader holding a stale
+// registration does: a duplicate-uuid response and nothing else.
+type duplicateUUIDMaster struct {
+	master_pb.UnimplementedSeaweedServer
+}
+
+func (m *duplicateUUIDMaster) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServer) error {
+	for {
+		if _, err := stream.Recv(); err != nil {
+			return err
+		}
+		if err := stream.Send(&master_pb.HeartbeatResponse{DuplicatedUuids: []string{"uuid-1"}}); err != nil {
+			return err
+		}
+	}
+}
+
+// A leader that rejects the registration over duplicate uuids still answers
+// lookups, so it is published before the retry error goes out.
+func TestHeartbeatPublishesLeaderBeforeDuplicateUuidError(t *testing.T) {
+	masterAddr := startFakeMasterServerForLeaderLookup(t, &duplicateUUIDMaster{})
+	vs := newHeartbeatTestVolumeServer(t)
+	vs.setCurrentMaster("previous.example.com:9333.19333")
+
+	_, err := vs.doHeartbeatWithRetry(masterAddr, vs.grpcDialOption, time.Hour, 0)
+	if err == nil || !strings.Contains(err.Error(), "duplicate UUIDs") {
+		t.Fatalf("err = %v, want duplicate UUIDs", err)
+	}
+	if got := vs.getCurrentMaster(); got != masterAddr {
+		t.Fatalf("currentMaster = %q, want the answering leader %q", got, masterAddr)
+	}
+}
+
+// After the heartbeat loses its master, lookups move to the leader the other
+// masters report instead of dialing the lost one until a new stream answers.
+func TestRefreshLeaderFromPeersPublishesReportedLeader(t *testing.T) {
+	const leader = pb.ServerAddress("leader.example.com:9333.19333")
+	peer := startFakeMasterServerForLeaderLookup(t, &fakeLeaderConfigServer{leader: string(leader)})
+	vs := &VolumeServer{
+		SeedMasterNodes: []pb.ServerAddress{peer},
+		grpcDialOption:  grpc.WithTransportCredentials(insecure.NewCredentials()),
+		store:           &storage.Store{Ip: "127.0.0.1", Port: 8080},
+	}
+	vs.store.MasterFn = vs.getCurrentMaster
+	vs.setCurrentMaster("lost.example.com:9333.19333")
+
+	vs.refreshLeaderFromPeers(5 * time.Second)
+
+	if got := vs.getCurrentMaster(); got != leader {
+		t.Fatalf("currentMaster = %q, want reported leader %q", string(got), string(leader))
+	}
+	if got := vs.store.MasterFn(); got != leader {
+		t.Fatalf("store master = %q, want %q", string(got), string(leader))
+	}
+}
+
+// Mid-election no master names a leader. The lost master is dead or now a
+// follower that would answer lookups from its own cache, so none is published.
+func TestRefreshLeaderFromPeersClearsLeaderWithoutAnswer(t *testing.T) {
+	peer := startFakeMasterServerForLeaderLookup(t, &fakeLeaderConfigServer{})
+	vs := &VolumeServer{
+		SeedMasterNodes: []pb.ServerAddress{peer},
+		grpcDialOption:  grpc.WithTransportCredentials(insecure.NewCredentials()),
+		store:           &storage.Store{Ip: "127.0.0.1", Port: 8080},
+	}
+	vs.store.MasterFn = vs.getCurrentMaster
+	vs.setCurrentMaster("lost.example.com:9333.19333")
+
+	vs.refreshLeaderFromPeers(5 * time.Second)
+
+	if got := vs.getCurrentMaster(); got != "" {
+		t.Fatalf("currentMaster = %q, want none", string(got))
+	}
+	if got := vs.store.MasterFn(); got != "" {
+		t.Fatalf("store master = %q, want none", string(got))
+	}
+}

weed/server/volume_server.go

@@ -148,6 +148,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 	vs.checkWithMaster()
 
 	vs.store = storage.NewStore(vs.grpcDialOption, ip, port, grpcPort, publicUrl, id, folders, maxCounts, minFreeSpaces, idxFolder, vs.needleMapKind, diskTypes, diskTags, ldbTimeout, diskProbeConfig)
+	vs.store.MasterFn = vs.getCurrentMaster
 	vs.guard = security.NewGuard(whiteList, signingKey, expiresAfterSec, readSigningKey, readExpiresAfterSec)
 
 	handleStaticResources(adminMux)

weed/storage/erasure_coding/distribution/README.md

@@ -30,8 +30,8 @@ fmt.Println(dist.Summary())
 // Replication: replication=110 (DCs:2, Racks/DC:2, Nodes/Rack:1)
 // Distribution Plan:
 //   Data Centers: 2 (target 7 shards each, max 9)
-//   Racks per DC: 2 (target 4 shards each, max 6)
-//   Nodes per Rack: 1 (target 4 shards each, max 6)
+//   Racks per DC: 2 (target 4 shards each, max 4)
+//   Nodes per Rack: 1 (target 4 shards each, max 4)
 ```
 
 ### Custom EC Ratios (seaweed-enterprise)

weed/storage/erasure_coding/distribution/analysis.go

@@ -1,5 +1,103 @@
 package distribution
 
+// NewTopologyAnalysis creates an empty topology analysis.
+func NewTopologyAnalysis() *TopologyAnalysis {
+	return &TopologyAnalysis{
+		ShardsByDC:   make(map[string]int),
+		ShardsByRack: make(map[string]int),
+		ShardsByNode: make(map[string]int),
+		DCToShards:   make(map[string][]int),
+		RackToShards: make(map[string][]int),
+		NodeToShards: make(map[string][]int),
+		DCToRacks:    make(map[string][]string),
+		RackToNodes:  make(map[string][]*TopologyNode),
+		AllNodes:     make(map[string]*TopologyNode),
+	}
+}
+
+// AddNode registers a node in the topology, including any shards it already
+// holds for the volume being analyzed. A node id may only register once;
+// registering it again updates its capacity and adds any new shards. The
+// first registration wins for the node's place in the topology.
+func (a *TopologyAnalysis) AddNode(node *TopologyNode) {
+	registered, ok := a.AllNodes[node.NodeID]
+	if !ok {
+		registered = node
+		a.AllNodes[node.NodeID] = node
+		a.RackToNodes[rackKey(node.DataCenter, node.Rack)] = append(a.RackToNodes[rackKey(node.DataCenter, node.Rack)], node)
+		if !contains(a.DCToRacks[node.DataCenter], node.Rack) {
+			a.DCToRacks[node.DataCenter] = append(a.DCToRacks[node.DataCenter], node.Rack)
+		}
+	} else {
+		registered.FreeSlots = node.FreeSlots
+		registered.TotalShards = node.TotalShards
+	}
+	seen := node.ShardIDs[:0]
+	for _, shardID := range node.ShardIDs {
+		if !contains(seen, shardID) {
+			seen = append(seen, shardID)
+		}
+	}
+	node.ShardIDs = seen
+	for _, shardID := range node.ShardIDs {
+		a.AddShardLocation(ShardLocation{
+			ShardID:    shardID,
+			NodeID:     registered.NodeID,
+			DataCenter: registered.DataCenter,
+			Rack:       registered.Rack,
+		})
+	}
+}
+
+// AddShardLocation records one shard's placement in the topology. Repeated
+// calls for a shard already known to its node are ignored, so callers may
+// either set ShardIDs on the TopologyNode or register each shard here. A
+// location naming an unregistered node registers that node so the rebalancer
+// can see the shard; for a registered node the node's own dc/rack win.
+func (a *TopologyAnalysis) AddShardLocation(loc ShardLocation) {
+	node, ok := a.AllNodes[loc.NodeID]
+	if !ok {
+		node = &TopologyNode{NodeID: loc.NodeID, DataCenter: loc.DataCenter, Rack: loc.Rack}
+		a.AddNode(node)
+	}
+	if contains(a.NodeToShards[loc.NodeID], loc.ShardID) {
+		return
+	}
+	a.ShardsByDC[node.DataCenter]++
+	a.ShardsByRack[rackKey(node.DataCenter, node.Rack)]++
+	a.ShardsByNode[loc.NodeID]++
+	a.DCToShards[node.DataCenter] = append(a.DCToShards[node.DataCenter], loc.ShardID)
+	a.RackToShards[rackKey(node.DataCenter, node.Rack)] = append(a.RackToShards[rackKey(node.DataCenter, node.Rack)], loc.ShardID)
+	a.NodeToShards[loc.NodeID] = append(a.NodeToShards[loc.NodeID], loc.ShardID)
+	a.TotalShards++
+	if !contains(node.ShardIDs, loc.ShardID) {
+		node.ShardIDs = append(node.ShardIDs, loc.ShardID)
+	}
+}
+
+// rackKey qualifies a rack name with its data center: rack names are only
+// unique within a DC, so two DCs can both hold a "rack1". NUL separates the
+// parts so a "/" inside either name cannot merge distinct racks.
+func rackKey(dc, rack string) string {
+	return dc + "\x00" + rack
+}
+
+// Finalize computes topology statistics after all nodes and shards are added.
+func (a *TopologyAnalysis) Finalize() {
+	a.TotalNodes = len(a.AllNodes)
+	a.TotalRacks = len(a.RackToNodes)
+	a.TotalDCs = len(a.DCToRacks)
+}
+
+func contains[T comparable](list []T, s T) bool {
+	for _, item := range list {
+		if item == s {
+			return true
+		}
+	}
+	return false
+}
+
 // ShardLocation represents where a shard is located in the topology
 type ShardLocation struct {
 	ShardID    int

weed/storage/erasure_coding/distribution/config.go

@@ -1,6 +1,14 @@
 // Package distribution provides EC shard distribution algorithms with configurable EC ratios.
 package distribution
 
+import (
+	"fmt"
+	"sort"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+)
+
 // ECConfig holds erasure coding configuration parameters.
 // This replaces hard-coded constants like DataShardsCount=10, ParityShardsCount=4.
 type ECConfig struct {
@@ -17,3 +25,76 @@ type ReplicationConfig struct {
 	// Original replication string (for logging/debugging)
 	Original string
 }
+
+// NewECConfig creates an erasure coding configuration with the given shard counts.
+func NewECConfig(dataShards, parityShards int) (ECConfig, error) {
+	if dataShards < 1 || parityShards < 1 {
+		return ECConfig{}, fmt.Errorf("data and parity shard counts must be positive, got %d+%d", dataShards, parityShards)
+	}
+	return ECConfig{DataShards: dataShards, ParityShards: parityShards}, nil
+}
+
+// DefaultECConfig returns the standard 10+4 erasure coding configuration.
+func DefaultECConfig() ECConfig {
+	return ECConfig{DataShards: erasure_coding.DataShardsCount, ParityShards: erasure_coding.ParityShardsCount}
+}
+
+// NewReplicationConfig parses a replication policy string like "110".
+func NewReplicationConfigFromString(s string) (ReplicationConfig, error) {
+	rp, err := super_block.NewReplicaPlacementFromString(s)
+	if err != nil {
+		return ReplicationConfig{}, err
+	}
+	return NewReplicationConfig(rp), nil
+}
+
+// NewReplicationConfig converts a replica placement into a replication config.
+func NewReplicationConfig(rp *super_block.ReplicaPlacement) ReplicationConfig {
+	if rp == nil {
+		rp = &super_block.ReplicaPlacement{}
+	}
+	return ReplicationConfig{
+		MinDataCenters:  rp.DiffDataCenterCount + 1,
+		MinRacksPerDC:   rp.DiffRackCount + 1,
+		MinNodesPerRack: rp.SameRackCount + 1,
+		Original:        rp.String(),
+	}
+}
+
+// TotalShards returns the number of shards each volume splits into.
+func (ec ECConfig) TotalShards() int {
+	return ec.DataShards + ec.ParityShards
+}
+
+// CanLoseShards returns how many shards of a volume can be missing before the
+// volume is no longer readable.
+func (ec ECConfig) CanLoseShards() int {
+	return ec.ParityShards
+}
+
+// IsDataShard reports whether the shard id is a data shard, which serves reads.
+func (ec ECConfig) IsDataShard(shardID int) bool {
+	return shardID >= 0 && shardID < ec.DataShards
+}
+
+// IsParityShard reports whether the shard id is a parity shard, which is only
+// used for reconstruction.
+func (ec ECConfig) IsParityShard(shardID int) bool {
+	return shardID >= ec.DataShards && shardID < ec.TotalShards()
+}
+
+// SortShardsDataFirst orders shard ids so data shards come before parity
+// shards, for initial placement.
+func (ec ECConfig) SortShardsDataFirst(shards []int) []int {
+	sorted := append([]int(nil), shards...)
+	sort.Ints(sorted)
+	return sorted
+}
+
+// SortShardsParityFirst orders shard ids so parity shards come before data
+// shards, for picking which shards to move away first when rebalancing.
+func (ec ECConfig) SortShardsParityFirst(shards []int) []int {
+	sorted := append([]int(nil), shards...)
+	sort.Sort(sort.Reverse(sort.IntSlice(sorted)))
+	return sorted
+}

weed/storage/erasure_coding/distribution/distribution.go

@@ -1,5 +1,10 @@
 package distribution
 
+import (
+	"fmt"
+	"strings"
+)
+
 // ECDistribution represents the target distribution of EC shards
 // based on EC configuration and replication policy.
 type ECDistribution struct {
@@ -20,3 +25,97 @@ type ECDistribution struct {
 	MaxShardsPerRack int
 	MaxShardsPerNode int
 }
+
+// CalculateDistribution computes the balanced shard counts each topology level
+// should hold for the given EC ratio spread over the minimum topology the
+// replication policy describes.
+func CalculateDistribution(ec ECConfig, rep ReplicationConfig) *ECDistribution {
+	total := ec.TotalShards()
+	rep.MinDataCenters = max(rep.MinDataCenters, 1)
+	rep.MinRacksPerDC = max(rep.MinRacksPerDC, 1)
+	rep.MinNodesPerRack = max(rep.MinNodesPerRack, 1)
+	dcUnits := rep.MinDataCenters
+	rackUnits := dcUnits * rep.MinRacksPerDC
+	nodeUnits := rackUnits * rep.MinNodesPerRack
+	return &ECDistribution{
+		ECConfig:            ec,
+		ReplicationConfig:   rep,
+		TargetShardsPerDC:   evenShare(total, dcUnits),
+		TargetShardsPerRack: evenShare(total, rackUnits),
+		TargetShardsPerNode: evenShare(total, nodeUnits),
+		MaxShardsPerDC:      maxShare(total, dcUnits, ec.CanLoseShards()),
+		MaxShardsPerRack:    maxShare(total, rackUnits, ec.CanLoseShards()),
+		MaxShardsPerNode:    maxShare(total, nodeUnits, ec.CanLoseShards()),
+	}
+}
+
+// evenShare is the balanced shard count per unit at one topology level.
+func evenShare(total, units int) int {
+	return ceilDivide(total, units)
+}
+
+// maxShare bounds how many shards a single failure domain may hold: the even
+// share plus the slack that would pad the total to the next multiple of the
+// unit count. When the even share is survivable on its own, the loss budget
+// caps the bound so the fault tolerance report stays true even at the max.
+func maxShare(total, units, lossBudget int) int {
+	if units < 1 {
+		units = 1
+	}
+	m := ceilDivide(total, units) + units - total%units
+	if m > total {
+		m = total
+	}
+	if evenShare(total, units) <= lossBudget && m > lossBudget {
+		m = lossBudget
+	}
+	return m
+}
+
+func ceilDivide(a, b int) int {
+	if b < 1 {
+		return a
+	}
+	return (a + b - 1) / b
+}
+
+// Summary describes the distribution plan in human-readable form.
+func (d *ECDistribution) Summary() string {
+	var sb strings.Builder
+	ec := d.ECConfig
+	rep := d.ReplicationConfig
+	fmt.Fprintf(&sb, "EC Configuration: %d+%d (total: %d, can lose: %d)\n",
+		ec.DataShards, ec.ParityShards, ec.TotalShards(), ec.CanLoseShards())
+	fmt.Fprintf(&sb, "Replication: replication=%s (DCs:%d, Racks/DC:%d, Nodes/Rack:%d)\n",
+		rep.Original, rep.MinDataCenters, rep.MinRacksPerDC, rep.MinNodesPerRack)
+	sb.WriteString("Distribution Plan:\n")
+	fmt.Fprintf(&sb, "  Data Centers: %d (target %d shards each, max %d)\n",
+		rep.MinDataCenters, d.TargetShardsPerDC, d.MaxShardsPerDC)
+	fmt.Fprintf(&sb, "  Racks per DC: %d (target %d shards each, max %d)\n",
+		rep.MinRacksPerDC, d.TargetShardsPerRack, d.MaxShardsPerRack)
+	fmt.Fprintf(&sb, "  Nodes per Rack: %d (target %d shards each, max %d)\n",
+		rep.MinNodesPerRack, d.TargetShardsPerNode, d.MaxShardsPerNode)
+	return sb.String()
+}
+
+// FaultToleranceAnalysis reports whether losing one unit at each topology
+// level still leaves enough shards to reconstruct the volume.
+func (d *ECDistribution) FaultToleranceAnalysis() string {
+	var sb strings.Builder
+	ec := d.ECConfig
+	fmt.Fprintf(&sb, "Fault Tolerance Analysis for %d+%d:\n", ec.DataShards, ec.ParityShards)
+	writeLevel := func(name string, target int) {
+		remaining := ec.TotalShards() - target
+		status := "SURVIVABLE ✓"
+		if remaining < ec.DataShards {
+			status = "NOT SURVIVABLE ✗"
+		}
+		fmt.Fprintf(&sb, "  %s Failure: %s\n", name, status)
+		fmt.Fprintf(&sb, "    - Losing one %s loses ~%d shards\n", name, target)
+		fmt.Fprintf(&sb, "    - Remaining: %d shards (need %d)\n", remaining, ec.DataShards)
+	}
+	writeLevel("DC", d.TargetShardsPerDC)
+	writeLevel("Rack", d.TargetShardsPerRack)
+	writeLevel("Node", d.TargetShardsPerNode)
+	return sb.String()
+}

weed/storage/erasure_coding/distribution/distribution_test.go

@@ -0,0 +1,456 @@
+package distribution
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDefaultECConfig(t *testing.T) {
+	ec := DefaultECConfig()
+	assert.Equal(t, 10, ec.DataShards)
+	assert.Equal(t, 4, ec.ParityShards)
+	assert.Equal(t, 14, ec.TotalShards())
+	assert.Equal(t, 4, ec.CanLoseShards())
+	assert.True(t, ec.IsDataShard(0))
+	assert.True(t, ec.IsDataShard(9))
+	assert.False(t, ec.IsDataShard(10))
+	assert.True(t, ec.IsParityShard(13))
+	assert.False(t, ec.IsParityShard(14))
+	assert.Equal(t, []int{0, 1, 10, 11}, ec.SortShardsDataFirst([]int{11, 1, 10, 0}))
+	assert.Equal(t, []int{13, 12, 2, 1}, ec.SortShardsParityFirst([]int{2, 13, 1, 12}))
+}
+
+func TestNewECConfig(t *testing.T) {
+	ec, err := NewECConfig(8, 4)
+	require.NoError(t, err)
+	assert.Equal(t, 12, ec.TotalShards())
+
+	for _, bad := range [][2]int{{0, 4}, {8, 0}, {-1, 4}} {
+		_, err := NewECConfig(bad[0], bad[1])
+		assert.Error(t, err, "config %v must be rejected", bad)
+	}
+}
+
+func TestNewReplicationConfigFromString(t *testing.T) {
+	rep, err := NewReplicationConfigFromString("110")
+	require.NoError(t, err)
+	assert.Equal(t, 2, rep.MinDataCenters)
+	assert.Equal(t, 2, rep.MinRacksPerDC)
+	assert.Equal(t, 1, rep.MinNodesPerRack)
+	assert.Equal(t, "110", rep.Original)
+
+	rep, err = NewReplicationConfigFromString("000")
+	require.NoError(t, err)
+	assert.Equal(t, 1, rep.MinDataCenters)
+	assert.Equal(t, 1, rep.MinRacksPerDC)
+	assert.Equal(t, 1, rep.MinNodesPerRack)
+
+	_, err = NewReplicationConfigFromString("abc")
+	assert.Error(t, err)
+}
+
+func TestCalculateDistribution(t *testing.T) {
+	rep, err := NewReplicationConfigFromString("110")
+	require.NoError(t, err)
+	dist := CalculateDistribution(DefaultECConfig(), rep)
+
+	assert.Equal(t, 7, dist.TargetShardsPerDC)
+	assert.Equal(t, 4, dist.TargetShardsPerRack)
+	assert.Equal(t, 4, dist.TargetShardsPerNode)
+	assert.Equal(t, 9, dist.MaxShardsPerDC)
+	assert.Equal(t, 4, dist.MaxShardsPerRack)
+	assert.Equal(t, 4, dist.MaxShardsPerNode)
+
+	expected := "EC Configuration: 10+4 (total: 14, can lose: 4)\n" +
+		"Replication: replication=110 (DCs:2, Racks/DC:2, Nodes/Rack:1)\n" +
+		"Distribution Plan:\n" +
+		"  Data Centers: 2 (target 7 shards each, max 9)\n" +
+		"  Racks per DC: 2 (target 4 shards each, max 4)\n" +
+		"  Nodes per Rack: 1 (target 4 shards each, max 4)\n"
+	assert.Equal(t, expected, dist.Summary())
+}
+
+// The reported max may not exceed the loss budget at a survivable level:
+// 8+4 over "200" lets a domain drift to 7 of 12 shards, but losing that domain
+// would leave 5 < 8 — capping the max at the budget keeps SURVIVABLE true.
+func TestCalculateDistributionMaxWithinLossBudget(t *testing.T) {
+	ec, err := NewECConfig(8, 4)
+	require.NoError(t, err)
+	rep, err := NewReplicationConfigFromString("200")
+	require.NoError(t, err)
+	dist := CalculateDistribution(ec, rep)
+
+	assert.Equal(t, 4, dist.TargetShardsPerDC)
+	assert.Equal(t, 4, dist.MaxShardsPerDC)
+	assert.Equal(t, 4, dist.MaxShardsPerRack)
+	assert.Equal(t, 4, dist.MaxShardsPerNode)
+
+	// a level whose even share already exceeds the budget keeps the slack bound
+	dist = CalculateDistribution(DefaultECConfig(), rep)
+	assert.Equal(t, 6, dist.MaxShardsPerDC) // even 5 > 4: DC loss is not survivable anyway
+}
+
+func TestCalculateDistributionSingleDomain(t *testing.T) {
+	rep, err := NewReplicationConfigFromString("000")
+	require.NoError(t, err)
+	dist := CalculateDistribution(DefaultECConfig(), rep)
+	assert.Equal(t, 14, dist.TargetShardsPerDC)
+	assert.Equal(t, 14, dist.MaxShardsPerDC)
+}
+
+func TestFaultToleranceAnalysis(t *testing.T) {
+	ec, err := NewECConfig(8, 4)
+	require.NoError(t, err)
+	rep, err := NewReplicationConfigFromString("200")
+	require.NoError(t, err)
+	dist := CalculateDistribution(ec, rep)
+
+	analysis := dist.FaultToleranceAnalysis()
+	assert.Contains(t, analysis, "Fault Tolerance Analysis for 8+4:")
+	assert.Contains(t, analysis, "DC Failure: SURVIVABLE")
+	assert.Contains(t, analysis, "- Losing one DC loses ~4 shards")
+	assert.Contains(t, analysis, "- Remaining: 8 shards (need 8)")
+}
+
+func TestFaultToleranceAnalysisNotSurvivable(t *testing.T) {
+	rep, err := NewReplicationConfigFromString("000")
+	require.NoError(t, err)
+	dist := CalculateDistribution(DefaultECConfig(), rep)
+	analysis := dist.FaultToleranceAnalysis()
+	assert.Contains(t, analysis, "DC Failure: NOT SURVIVABLE")
+	assert.Contains(t, analysis, "Rack Failure: NOT SURVIVABLE")
+	assert.Contains(t, analysis, "Node Failure: NOT SURVIVABLE")
+}
+
+func makeNode(id, dc, rack string, freeSlots int, shards ...int) *TopologyNode {
+	return &TopologyNode{NodeID: id, DataCenter: dc, Rack: rack, FreeSlots: freeSlots, ShardIDs: shards}
+}
+
+func TestTopologyAnalysis(t *testing.T) {
+	a := NewTopologyAnalysis()
+	a.AddNode(makeNode("n1", "dc1", "r1", 5, 0, 1))
+	a.AddNode(makeNode("n2", "dc1", "r1", 5, 2))
+	a.AddNode(makeNode("n3", "dc2", "r2", 5))
+	a.Finalize()
+
+	assert.Equal(t, 3, a.TotalShards)
+	assert.Equal(t, 3, a.TotalNodes)
+	assert.Equal(t, 2, a.TotalRacks)
+	assert.Equal(t, 2, a.TotalDCs)
+	assert.Equal(t, 3, a.ShardsByDC["dc1"])
+	assert.Equal(t, 0, a.ShardsByDC["dc2"])
+	assert.Equal(t, []string{"r1"}, a.DCToRacks["dc1"])
+	assert.Equal(t, []int{0, 1, 2}, a.RackToShards[rackKey("dc1", "r1")])
+}
+
+// A node registered twice must still count once in rack and shard totals.
+func TestTopologyAnalysisDuplicateNode(t *testing.T) {
+	a := NewTopologyAnalysis()
+	a.AddNode(makeNode("n1", "dc1", "r1", 5, 0, 1))
+	a.AddNode(makeNode("n1", "dc1", "r1", 5, 0, 1))
+	a.Finalize()
+
+	assert.Equal(t, 1, a.TotalNodes)
+	assert.Len(t, a.RackToNodes[rackKey("dc1", "r1")], 1)
+	assert.Equal(t, 2, a.ShardsByNode["n1"])
+}
+
+// A node auto-registered by a shard location gets its real capacity when the
+// node itself registers later, plus any shards the location did not mention.
+func TestTopologyAnalysisNodeCapacityLateRegistration(t *testing.T) {
+	a := NewTopologyAnalysis()
+	a.AddShardLocation(ShardLocation{ShardID: 0, NodeID: "n1", DataCenter: "dc1", Rack: "r1"})
+	a.AddNode(makeNode("n1", "dc1", "r1", 10, 1))
+	a.Finalize()
+
+	assert.Equal(t, 10, a.AllNodes["n1"].FreeSlots)
+	assert.Equal(t, []int{0, 1}, a.AllNodes["n1"].ShardIDs)
+	assert.Equal(t, 2, a.ShardsByNode["n1"])
+	assert.Len(t, a.RackToNodes[rackKey("dc1", "r1")], 1)
+}
+
+// Shards may be registered either through TopologyNode.ShardIDs or through
+// AddShardLocation — both must leave the node holding the shards.
+func TestTopologyAnalysisAddShardLocation(t *testing.T) {
+	a := NewTopologyAnalysis()
+	a.AddNode(makeNode("n1", "dc1", "r1", 5))
+	a.AddShardLocation(ShardLocation{ShardID: 0, NodeID: "n1", DataCenter: "dc1", Rack: "r1"})
+	a.AddShardLocation(ShardLocation{ShardID: 1, NodeID: "n1", DataCenter: "dc1", Rack: "r1"})
+	a.AddShardLocation(ShardLocation{ShardID: 1, NodeID: "n1", DataCenter: "dc1", Rack: "r1"})
+	a.Finalize()
+
+	assert.Equal(t, 2, a.TotalShards)
+	assert.Equal(t, 2, a.ShardsByNode["n1"])
+	assert.Equal(t, []int{0, 1}, a.AllNodes["n1"].ShardIDs)
+}
+
+// Rack names are only unique within a DC, so dc1/rack1 and dc2/rack1 are
+// distinct domains.
+func TestTopologyAnalysisSameRackNameAcrossDCs(t *testing.T) {
+	a := NewTopologyAnalysis()
+	a.AddNode(makeNode("n1", "dc1", "rack1", 5, 0, 1))
+	a.AddNode(makeNode("n2", "dc2", "rack1", 5, 2))
+	a.Finalize()
+
+	assert.Equal(t, 2, a.TotalDCs)
+	assert.Equal(t, 2, a.TotalRacks)
+	assert.Equal(t, 2, a.ShardsByRack[rackKey("dc1", "rack1")])
+	assert.Equal(t, 1, a.ShardsByRack[rackKey("dc2", "rack1")])
+}
+
+// A "/" inside a data-center or rack name must not merge distinct racks.
+func TestTopologyAnalysisSlashInNames(t *testing.T) {
+	a := NewTopologyAnalysis()
+	a.AddNode(makeNode("n1", "a", "b/c", 5, 0))
+	a.AddNode(makeNode("n2", "a/b", "c", 5, 1))
+	a.Finalize()
+
+	assert.Equal(t, 2, a.TotalRacks)
+	assert.Equal(t, 1, a.ShardsByRack[rackKey("a", "b/c")])
+	assert.Equal(t, 1, a.ShardsByRack[rackKey("a/b", "c")])
+}
+
+func TestPlanRebalanceAcrossDCs(t *testing.T) {
+	a := NewTopologyAnalysis()
+	a.AddNode(makeNode("a1", "dcA", "rA1", 10, 0, 1, 2, 3))
+	a.AddNode(makeNode("a2", "dcA", "rA2", 10, 4, 5, 6, 7))
+	a.AddNode(makeNode("a3", "dcA", "rA3", 10, 8, 9, 10, 11))
+	a.AddNode(makeNode("a4", "dcA", "rA4", 10, 12, 13))
+	a.AddNode(makeNode("b1", "dcB", "rB1", 10))
+	a.AddNode(makeNode("b2", "dcB", "rB2", 10))
+	a.Finalize()
+
+	rep, err := NewReplicationConfigFromString("110")
+	require.NoError(t, err)
+	plan, err := NewRebalancer(DefaultECConfig(), rep).PlanRebalance(a)
+	require.NoError(t, err)
+
+	assert.Equal(t, 7, plan.MovesAcrossDC)
+	assert.Equal(t, 7, plan.Analysis.ShardsByDC["dcA"])
+	assert.Equal(t, 7, plan.Analysis.ShardsByDC["dcB"])
+
+	// the given analysis still describes the un-moved layout
+	assert.Equal(t, 14, a.ShardsByDC["dcA"])
+	assert.Equal(t, 0, a.ShardsByDC["dcB"])
+	assert.Equal(t, []int{0, 1, 2, 3}, a.AllNodes["a1"].ShardIDs)
+
+	// parity shards (10-13) move before data shards
+	assert.True(t, plan.Moves[0].ShardID >= 10)
+	assert.Equal(t, "b1", plan.Moves[0].DestNode.NodeID)
+}
+
+func TestPlanRebalanceAcrossRacks(t *testing.T) {
+	a := NewTopologyAnalysis()
+	a.AddNode(makeNode("n1", "dc1", "r1", 10, 0, 1, 2, 3))
+	a.AddNode(makeNode("n2", "dc1", "r2", 10))
+	a.Finalize()
+
+	rep, err := NewReplicationConfigFromString("110")
+	require.NoError(t, err)
+	plan, err := NewRebalancer(DefaultECConfig(), rep).PlanRebalance(a)
+	require.NoError(t, err)
+
+	assert.Equal(t, 2, plan.TotalMoves)
+	assert.Equal(t, 2, plan.MovesAcrossRack)
+	assert.Equal(t, 2, plan.Analysis.ShardsByRack[rackKey("dc1", "r1")])
+	assert.Equal(t, 2, plan.Analysis.ShardsByRack[rackKey("dc1", "r2")])
+	// parity-first ordering: shards 3 then 2 move out of r1
+	assert.Equal(t, 3, plan.Moves[0].ShardID)
+	assert.Equal(t, 2, plan.Moves[1].ShardID)
+}
+
+func TestPlanRebalanceWithinRack(t *testing.T) {
+	a := NewTopologyAnalysis()
+	a.AddNode(makeNode("n1", "dc1", "r1", 10, 0, 1, 2, 3))
+	a.AddNode(makeNode("n2", "dc1", "r1", 10))
+	a.Finalize()
+
+	rep, err := NewReplicationConfigFromString("001")
+	require.NoError(t, err)
+	plan, err := NewRebalancer(DefaultECConfig(), rep).PlanRebalance(a)
+	require.NoError(t, err)
+
+	assert.Equal(t, 2, plan.TotalMoves)
+	assert.Equal(t, 2, plan.MovesWithinRack)
+	assert.Equal(t, 2, plan.Analysis.ShardsByNode["n1"])
+	assert.Equal(t, 2, plan.Analysis.ShardsByNode["n2"])
+	assert.Equal(t, []int{3, 2}, plan.Analysis.AllNodes["n2"].ShardIDs)
+	assert.Equal(t, []int{0, 1, 2, 3}, a.AllNodes["n1"].ShardIDs)
+}
+
+func TestPlanRebalanceRespectsFreeSlots(t *testing.T) {
+	a := NewTopologyAnalysis()
+	a.AddNode(makeNode("n1", "dc1", "r1", 10, 0, 1, 2, 3))
+	a.AddNode(makeNode("n2", "dc1", "r1", 1))
+	a.Finalize()
+
+	plan, err := NewRebalancer(DefaultECConfig(), ReplicationConfig{MinDataCenters: 1, MinRacksPerDC: 1, MinNodesPerRack: 2}).PlanRebalance(a)
+	require.NoError(t, err)
+
+	assert.Equal(t, 1, plan.TotalMoves, "only one slot is free on the destination")
+	assert.Equal(t, 3, plan.Analysis.ShardsByNode["n1"])
+	assert.Equal(t, 1, plan.Analysis.ShardsByNode["n2"])
+	assert.Equal(t, 0, plan.Analysis.AllNodes["n2"].FreeSlots)
+	assert.Equal(t, 1, a.AllNodes["n2"].FreeSlots)
+}
+
+func TestPlanRebalanceNoopWhenBalanced(t *testing.T) {
+	a := NewTopologyAnalysis()
+	a.AddNode(makeNode("n1", "dc1", "r1", 10, 0, 1))
+	a.AddNode(makeNode("n2", "dc1", "r1", 10, 2, 3))
+	a.Finalize()
+
+	plan, err := NewRebalancer(DefaultECConfig(), ReplicationConfig{MinDataCenters: 1, MinRacksPerDC: 1, MinNodesPerRack: 2}).PlanRebalance(a)
+	require.NoError(t, err)
+	assert.Equal(t, 0, plan.TotalMoves)
+}
+
+func TestPlanRebalanceNilAnalysis(t *testing.T) {
+	_, err := NewRebalancer(DefaultECConfig(), ReplicationConfig{}).PlanRebalance(nil)
+	assert.Error(t, err)
+}
+
+// Node balancing inside a rack must aim for the rack's count after
+// DC- and rack-level moves, not the count the rack started with.
+func TestPlanRebalanceNodeTargetAfterMoves(t *testing.T) {
+	a := NewTopologyAnalysis()
+	a.AddNode(makeNode("n1", "dc1", "r1", 10, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13))
+	a.AddNode(makeNode("n2", "dc1", "r1", 10))
+	a.AddNode(makeNode("n3", "dc1", "r2", 10))
+	a.Finalize()
+
+	rep, err := NewReplicationConfigFromString("010")
+	require.NoError(t, err)
+	plan, err := NewRebalancer(DefaultECConfig(), rep).PlanRebalance(a)
+	require.NoError(t, err)
+
+	assert.Equal(t, 7, plan.Analysis.ShardsByRack[rackKey("dc1", "r1")])
+	assert.Equal(t, 7, plan.Analysis.ShardsByRack[rackKey("dc1", "r2")])
+	assert.Equal(t, 4, plan.Analysis.ShardsByNode["n1"])
+	assert.Equal(t, 3, plan.Analysis.ShardsByNode["n2"])
+	assert.Equal(t, []int{0, 1, 2, 3}, plan.Analysis.AllNodes["n1"].ShardIDs)
+	assert.Equal(t, []int{6, 5, 4}, plan.Analysis.AllNodes["n2"].ShardIDs)
+}
+
+// A shard that already has a copy anywhere in the destination domain must not
+// be moved there again — the second copy adds no redundancy.
+func TestPlanRebalanceNoDuplicateShardInDomain(t *testing.T) {
+	a := NewTopologyAnalysis()
+	a.AddNode(makeNode("a1", "dcA", "rA1", 10, 0, 1, 2, 3))
+	a.AddNode(makeNode("b1", "dcB", "rB1", 10, 3))
+	a.AddNode(makeNode("b2", "dcB", "rB2", 10))
+	a.Finalize()
+
+	rep, err := NewReplicationConfigFromString("100")
+	require.NoError(t, err)
+	plan, err := NewRebalancer(DefaultECConfig(), rep).PlanRebalance(a)
+	require.NoError(t, err)
+
+	// dcA holds 4, dcB holds 1, target 3 each: one move, and it cannot be shard 3
+	require.Len(t, plan.Moves, 1)
+	assert.Equal(t, 2, plan.Moves[0].ShardID)
+	assert.Equal(t, []int{2}, plan.Analysis.AllNodes["b2"].ShardIDs)
+	assert.Equal(t, []int{3}, plan.Analysis.AllNodes["b1"].ShardIDs)
+}
+
+// Shard locations may be registered without registering the node first; the
+// rebalancer must still see the shard, counted under the node's own dc/rack.
+func TestPlanRebalanceShardForUnregisteredNode(t *testing.T) {
+	a := NewTopologyAnalysis()
+	a.AddShardLocation(ShardLocation{ShardID: 0, NodeID: "n1", DataCenter: "dc1", Rack: "r1"})
+	a.AddShardLocation(ShardLocation{ShardID: 1, NodeID: "n1", DataCenter: "dc1", Rack: "r1"})
+	a.AddNode(makeNode("n2", "dc1", "r1", 10))
+	a.Finalize()
+
+	assert.Equal(t, 1, a.TotalDCs)
+	assert.Equal(t, 2, a.ShardsByNode["n1"])
+	assert.Equal(t, []int{0, 1}, a.AllNodes["n1"].ShardIDs)
+
+	plan, err := NewRebalancer(DefaultECConfig(), ReplicationConfig{MinDataCenters: 1, MinRacksPerDC: 1, MinNodesPerRack: 2}).PlanRebalance(a)
+	require.NoError(t, err)
+	assert.Equal(t, 1, plan.TotalMoves)
+	assert.Equal(t, 1, plan.Analysis.ShardsByNode["n1"])
+	assert.Equal(t, 1, plan.Analysis.ShardsByNode["n2"])
+}
+
+// When the most overloaded domain can only offer duplicates, balancing must
+// fall through to the next overloaded source rather than stop.
+func TestPlanRebalanceSkipsBlockedSource(t *testing.T) {
+	a := NewTopologyAnalysis()
+	a.AddNode(makeNode("a1", "dcA", "rA", 10, 0, 1, 2, 3))
+	a.AddNode(makeNode("a2", "dcA", "rA", 10, 0, 1, 2, 3))
+	a.AddNode(makeNode("a3", "dcA", "rA", 10, 0, 1, 2, 3))
+	a.AddNode(makeNode("b1", "dcB", "rB", 10, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13))
+	a.AddNode(makeNode("c1", "dcC", "rC", 10, 0, 1, 2, 3))
+	a.AddNode(makeNode("c2", "dcC", "rC", 10))
+	a.AddNode(makeNode("d1", "dcD", "rD", 10, 0, 1, 2, 3))
+	a.AddNode(makeNode("d2", "dcD", "rD", 10))
+	a.Finalize()
+
+	plan, err := NewRebalancer(DefaultECConfig(), ReplicationConfig{MinDataCenters: 4, MinRacksPerDC: 1, MinNodesPerRack: 1}).PlanRebalance(a)
+	require.NoError(t, err)
+
+	// dcA is stuck on duplicates (target 8), but dcB still sheds two shards
+	assert.Equal(t, 12, plan.Analysis.ShardsByDC["dcA"])
+	assert.Equal(t, 8, plan.Analysis.ShardsByDC["dcB"])
+	assert.Equal(t, 5, plan.Analysis.ShardsByDC["dcC"])
+	assert.Equal(t, 5, plan.Analysis.ShardsByDC["dcD"])
+}
+
+// A shard that dc-level balancing lands on a node can be shed again at rack or
+// node level; its moves must collapse to one transfer to the final holder.
+func TestPlanRebalanceCollapsesChainedMoves(t *testing.T) {
+	a := NewTopologyAnalysis()
+	a.AddNode(makeNode("a1", "dcA", "rA", 20, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13))
+	a.AddNode(makeNode("b1", "dcB", "rB1", 20))
+	a.AddNode(makeNode("c1", "dcB", "rB2", 20))
+	a.AddNode(makeNode("c2", "dcB", "rB2", 20))
+	a.AddNode(makeNode("c3", "dcB", "rB2", 20))
+	a.Finalize()
+
+	rep, err := NewReplicationConfigFromString("100")
+	require.NoError(t, err)
+	plan, err := NewRebalancer(DefaultECConfig(), rep).PlanRebalance(a)
+	require.NoError(t, err)
+
+	moved := make(map[int]string)
+	for _, m := range plan.Moves {
+		_, twice := moved[m.ShardID]
+		assert.False(t, twice, "shard %d moves more than once", m.ShardID)
+		moved[m.ShardID] = m.DestNode.NodeID
+		assert.Contains(t, m.DestNode.ShardIDs, m.ShardID)
+	}
+	assert.Equal(t, 7, plan.TotalMoves)
+	assert.Equal(t, 7, plan.MovesAcrossDC)
+	// shard 12 reached b1 through c1 in simulation; the plan moves it directly
+	assert.Equal(t, "b1", moved[12])
+	assert.Equal(t, []int{13, 9, 12}, plan.Analysis.AllNodes["b1"].ShardIDs)
+}
+
+func TestPlanRebalanceNoDuplicateShardOnNode(t *testing.T) {
+	a := NewTopologyAnalysis()
+	a.AddNode(makeNode("n1", "dc1", "r1", 10, 0, 1, 2, 3))
+	a.AddNode(makeNode("n2", "dc1", "r1", 10, 3))
+	a.Finalize()
+
+	plan, err := NewRebalancer(DefaultECConfig(), ReplicationConfig{MinDataCenters: 1, MinRacksPerDC: 1, MinNodesPerRack: 2}).PlanRebalance(a)
+	require.NoError(t, err)
+
+	// n2 already holds 3, so the balancing move brings a different shard
+	assert.Equal(t, 3, plan.Analysis.ShardsByNode["n1"])
+	assert.Equal(t, 2, plan.Analysis.ShardsByNode["n2"])
+	assert.Equal(t, []int{3, 2}, plan.Analysis.AllNodes["n2"].ShardIDs)
+	for _, move := range plan.Moves {
+		assert.NotContains(t, move.DestNode.ShardIDs[:len(move.DestNode.ShardIDs)-1], move.ShardID,
+			"shard %d must not be moved onto a node already holding it", move.ShardID)
+	}
+}
+
+func TestCalculateDistributionZeroReplication(t *testing.T) {
+	dist := CalculateDistribution(DefaultECConfig(), ReplicationConfig{})
+	assert.Equal(t, 14, dist.TargetShardsPerDC)
+	assert.Contains(t, dist.Summary(), "Data Centers: 1")
+	assert.NotPanics(t, func() { dist.FaultToleranceAnalysis() })
+}

weed/storage/erasure_coding/distribution/rebalancer.go

@@ -1,5 +1,10 @@
 package distribution
 
+import (
+	"fmt"
+	"sort"
+)
+
 // ShardMove represents a planned shard move
 type ShardMove struct {
 	ShardID    int
@@ -12,7 +17,7 @@ type ShardMove struct {
 type RebalancePlan struct {
 	Moves        []ShardMove
 	Distribution *ECDistribution
-	Analysis     *TopologyAnalysis
+	Analysis     *TopologyAnalysis // simulated layout once every move applies; move endpoints reference its nodes
 
 	// Statistics
 	TotalMoves      int
@@ -26,3 +31,274 @@ type Rebalancer struct {
 	ecConfig  ECConfig
 	repConfig ReplicationConfig
 }
+
+// NewRebalancer creates a rebalancer for the given EC ratio and replication policy.
+func NewRebalancer(ec ECConfig, rep ReplicationConfig) *Rebalancer {
+	return &Rebalancer{ecConfig: ec, repConfig: rep}
+}
+
+// PlanRebalance generates shard moves that bring the volume's shard layout to
+// the balanced distribution: first across data centers, then racks within each
+// data center, then nodes within each rack. Moves stop when source and
+// destination both reach their even share, or when destination nodes have no
+// free slots. The moves are planned on a copy of the topology, so the given
+// analysis still describes the cluster as it is; plan.Analysis shows the
+// layout the moves would produce.
+func (r *Rebalancer) PlanRebalance(analysis *TopologyAnalysis) (*RebalancePlan, error) {
+	if analysis == nil {
+		return nil, fmt.Errorf("cannot plan rebalancing without a topology analysis")
+	}
+	sim := NewTopologyAnalysis()
+	for _, id := range sortedKeys(analysis.AllNodes) {
+		node := *analysis.AllNodes[id]
+		node.ShardIDs = append([]int(nil), node.ShardIDs...)
+		sim.AddNode(&node)
+	}
+	sim.Finalize()
+	plan := &RebalancePlan{
+		Distribution: CalculateDistribution(r.ecConfig, r.repConfig),
+		Analysis:     sim,
+	}
+
+	recordMove := func(shardID int, src, dst *TopologyNode, level, domain string) {
+		src.ShardIDs = removeShard(src.ShardIDs, shardID)
+		dst.ShardIDs = append(dst.ShardIDs, shardID)
+		src.FreeSlots++
+		dst.FreeSlots--
+		sim.ShardsByDC[src.DataCenter]--
+		sim.ShardsByDC[dst.DataCenter]++
+		sim.ShardsByRack[rackKey(src.DataCenter, src.Rack)]--
+		sim.ShardsByRack[rackKey(dst.DataCenter, dst.Rack)]++
+		sim.ShardsByNode[src.NodeID]--
+		sim.ShardsByNode[dst.NodeID]++
+		sim.DCToShards[src.DataCenter] = removeShard(sim.DCToShards[src.DataCenter], shardID)
+		sim.DCToShards[dst.DataCenter] = append(sim.DCToShards[dst.DataCenter], shardID)
+		sim.RackToShards[rackKey(src.DataCenter, src.Rack)] = removeShard(sim.RackToShards[rackKey(src.DataCenter, src.Rack)], shardID)
+		sim.RackToShards[rackKey(dst.DataCenter, dst.Rack)] = append(sim.RackToShards[rackKey(dst.DataCenter, dst.Rack)], shardID)
+		sim.NodeToShards[src.NodeID] = removeShard(sim.NodeToShards[src.NodeID], shardID)
+		sim.NodeToShards[dst.NodeID] = append(sim.NodeToShards[dst.NodeID], shardID)
+		plan.Moves = append(plan.Moves, ShardMove{
+			ShardID:    shardID,
+			SourceNode: src,
+			DestNode:   dst,
+			Reason:     fmt.Sprintf("%s balance: %s above target", level, domain),
+		})
+	}
+
+	moveOut := func(level, srcDomain string, srcNodes, dstNodes []*TopologyNode) bool {
+		var shardIDs []int
+		for _, n := range srcNodes {
+			shardIDs = append(shardIDs, n.ShardIDs...)
+		}
+		for _, shardID := range r.ecConfig.SortShardsParityFirst(shardIDs) {
+			var src *TopologyNode
+			for _, n := range srcNodes {
+				if contains(n.ShardIDs, shardID) {
+					src = n
+					break
+				}
+			}
+			if src == nil {
+				continue
+			}
+			// a shard already inside the destination domain gains nothing there
+			duplicate := false
+			for _, n := range dstNodes {
+				if contains(n.ShardIDs, shardID) {
+					duplicate = true
+					break
+				}
+			}
+			if duplicate {
+				continue
+			}
+			var dst *TopologyNode
+			for _, n := range dstNodes {
+				if n.FreeSlots > 0 && (dst == nil || nodeLess(n, dst)) {
+					dst = n
+				}
+			}
+			if dst == nil {
+				continue
+			}
+			recordMove(shardID, src, dst, level, srcDomain)
+			return true
+		}
+		return false
+	}
+
+	// balance moves shards between domains of one level until every domain
+	// holds at most its even share or no destination can take a shard. A
+	// source that cannot shed any shard is set aside rather than stopping
+	// the level, so other overloaded sources still get to move.
+	balance := func(level string, domains []string, countOf func(string) int, target int, nodesOf func(string) []*TopologyNode, dstDomains func(string) []string) {
+		blocked := make(map[string]bool)
+		for {
+			src := ""
+			srcFound := false
+			for _, d := range domains {
+				if blocked[d] {
+					continue
+				}
+				if countOf(d) > target && (!srcFound || countOf(d) > countOf(src)) {
+					src, srcFound = d, true
+				}
+			}
+			if !srcFound {
+				return
+			}
+			dsts := append([]string(nil), dstDomains(src)...)
+			sort.Slice(dsts, func(i, j int) bool {
+				if countOf(dsts[i]) != countOf(dsts[j]) {
+					return countOf(dsts[i]) < countOf(dsts[j])
+				}
+				return dsts[i] < dsts[j]
+			})
+			moved := false
+			for _, dst := range dsts {
+				if countOf(dst) < target && moveOut(level, src, nodesOf(src), nodesOf(dst)) {
+					moved = true
+					break
+				}
+			}
+			if !moved {
+				blocked[src] = true
+			}
+		}
+	}
+
+	dcList := sortedKeys(sim.DCToRacks)
+	dcRackKeys := func(dc string) []string {
+		racks := sim.DCToRacks[dc]
+		keys := make([]string, 0, len(racks))
+		for _, rack := range racks {
+			keys = append(keys, rackKey(dc, rack))
+		}
+		return keys
+	}
+	totalShards := 0
+	for _, dc := range dcList {
+		totalShards += shardCount(dcRackKeys(dc), sim.RackToNodes)
+	}
+	dcCount := func(dc string) int { return shardCount(dcRackKeys(dc), sim.RackToNodes) }
+	balance("dc", dcList, dcCount, ceilDivide(totalShards, len(dcList)),
+		func(dc string) []*TopologyNode { return nodesOfRacks(dcRackKeys(dc), sim.RackToNodes) },
+		func(string) []string { return dcList })
+
+	for _, dc := range dcList {
+		racks := sim.DCToRacks[dc]
+		rackCount := func(rack string) int { return shardCount([]string{rackKey(dc, rack)}, sim.RackToNodes) }
+		balance("rack", racks, rackCount, ceilDivide(dcCount(dc), len(racks)),
+			func(rack string) []*TopologyNode { return sim.RackToNodes[rackKey(dc, rack)] },
+			func(string) []string { return racks })
+	}
+
+	for _, rack := range sortedKeys(sim.RackToNodes) {
+		nodes := sim.RackToNodes[rack]
+		nodeCount := func(nodeID string) int {
+			for _, n := range nodes {
+				if n.NodeID == nodeID {
+					return len(n.ShardIDs)
+				}
+			}
+			return 0
+		}
+		nodeIDs := make([]string, 0, len(nodes))
+		for _, n := range nodes {
+			nodeIDs = append(nodeIDs, n.NodeID)
+		}
+		sort.Strings(nodeIDs)
+		balance("node", nodeIDs, nodeCount, ceilDivide(shardCount([]string{rack}, sim.RackToNodes), len(nodes)),
+			func(nodeID string) []*TopologyNode {
+				for _, n := range nodes {
+					if n.NodeID == nodeID {
+						return []*TopologyNode{n}
+					}
+				}
+				return nil
+			},
+			func(string) []string { return nodeIDs })
+	}
+
+	// A copy of a shard moves at most once per level, and each finer level only
+	// moves it deeper inside the domain the previous level sent it to, so its
+	// moves form a chain from the first source to the final destination.
+	// Collapse each chain into one transfer — intermediate hops would copy the
+	// shard over the network only to move it again. Chains are keyed by where
+	// the copy sits, so duplicate copies of a shard still move independently.
+	pending := make(map[ShardLocation]int)
+	moves := plan.Moves[:0]
+	for _, m := range plan.Moves {
+		key := ShardLocation{ShardID: m.ShardID, NodeID: m.SourceNode.NodeID}
+		if i, ok := pending[key]; ok {
+			delete(pending, key)
+			moves[i].DestNode = m.DestNode
+			pending[ShardLocation{ShardID: m.ShardID, NodeID: m.DestNode.NodeID}] = i
+			continue
+		}
+		pending[ShardLocation{ShardID: m.ShardID, NodeID: m.DestNode.NodeID}] = len(moves)
+		moves = append(moves, m)
+	}
+	plan.Moves = moves
+	for _, m := range moves {
+		switch {
+		case m.SourceNode.DataCenter != m.DestNode.DataCenter:
+			plan.MovesAcrossDC++
+		case m.SourceNode.Rack != m.DestNode.Rack:
+			plan.MovesAcrossRack++
+		default:
+			plan.MovesWithinRack++
+		}
+	}
+	plan.TotalMoves = len(moves)
+	return plan, nil
+}
+
+// nodeLess prefers the node holding fewer of the volume's shards, then the one
+// with more free slots, for a stable destination choice.
+func nodeLess(a, b *TopologyNode) bool {
+	if len(a.ShardIDs) != len(b.ShardIDs) {
+		return len(a.ShardIDs) < len(b.ShardIDs)
+	}
+	if a.FreeSlots != b.FreeSlots {
+		return a.FreeSlots > b.FreeSlots
+	}
+	return a.NodeID < b.NodeID
+}
+
+// shardCount sums the shards the given racks currently hold for the volume.
+func shardCount(racks []string, rackToNodes map[string][]*TopologyNode) int {
+	total := 0
+	for _, rack := range racks {
+		for _, n := range rackToNodes[rack] {
+			total += len(n.ShardIDs)
+		}
+	}
+	return total
+}
+
+func nodesOfRacks(racks []string, rackToNodes map[string][]*TopologyNode) []*TopologyNode {
+	var nodes []*TopologyNode
+	for _, rack := range racks {
+		nodes = append(nodes, rackToNodes[rack]...)
+	}
+	return nodes
+}
+
+func removeShard(shards []int, shardID int) []int {
+	for i, s := range shards {
+		if s == shardID {
+			return append(shards[:i], shards[i+1:]...)
+		}
+	}
+	return shards
+}
+
+func sortedKeys[V any](m map[string]V) []string {
+	keys := make([]string, 0, len(m))
+	for k := range m {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	return keys
+}

weed/storage/store.go

@@ -61,14 +61,25 @@ type ReadOption struct {
 	ReadBufferSize int
 }
 
+// masterAddress returns the master to ask for volume and shard locations, or
+// "" while no leader is known.
+func (s *Store) masterAddress() pb.ServerAddress {
+	if s.MasterFn == nil {
+		return ""
+	}
+	return s.MasterFn()
+}
+
 /*
  * A VolumeServer contains one Store
  */
 type Store struct {
-	MasterAddress       pb.ServerAddress
+	// MasterFn reports the master the volume server currently takes as leader.
+	// The heartbeat loop owns that answer; the store only reads it.
+	MasterFn            func() pb.ServerAddress
 	grpcDialOption      grpc.DialOption
-	volumeSizeLimit     uint64      // read from the master
-	preallocate         atomic.Bool // read from the master
+	volumeSizeLimit     atomic.Uint64 // read from the master
+	preallocate         atomic.Bool   // read from the master
 	Ip                  string
 	Port                int
 	GrpcPort            int
@@ -1302,11 +1313,11 @@ func (s *Store) ConfigureVolume(i needle.VolumeId, replication string) error {
 }
 
 func (s *Store) SetVolumeSizeLimit(x uint64) {
-	atomic.StoreUint64(&s.volumeSizeLimit, x)
+	s.volumeSizeLimit.Store(x)
 }
 
 func (s *Store) GetVolumeSizeLimit() uint64 {
-	return atomic.LoadUint64(&s.volumeSizeLimit)
+	return s.volumeSizeLimit.Load()
 }
 
 func (s *Store) SetPreallocate(x bool) {

weed/storage/store_ec.go

@@ -568,7 +568,7 @@ const ecIntervalReadConcurrency = 8
 
 func (s *Store) readEcShardIntervals(needleId types.NeedleId, ecVolume *erasure_coding.EcVolume, intervals []erasure_coding.Interval) (data []byte, is_deleted bool, err error) {
 	if err = s.cachedLookupEcShardLocations(ecVolume); err != nil {
-		return nil, false, fmt.Errorf("failed to locate shard via master grpc %s: %v", s.MasterAddress, err)
+		return nil, false, fmt.Errorf("failed to locate shard via master grpc %s: %v", s.masterAddress(), err)
 	}
 
 	var totalSize int
@@ -721,7 +721,7 @@ func (s *Store) cachedLookupEcShardLocations(ecVolume *erasure_coding.EcVolume)
 
 	glog.V(3).Infof("lookup and cache ec volume %d locations", ecVolume.VolumeId)
 
-	err = operation.WithMasterServerClient(context.Background(), false, s.MasterAddress, s.grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
+	err = operation.WithMasterServerClient(context.Background(), false, s.masterAddress(), s.grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
 		req := &master_pb.LookupEcVolumeRequest{
 			VolumeId: uint32(ecVolume.VolumeId),
 		}

weed/storage/store_ec_location_cache_test.go

@@ -78,7 +78,7 @@ func rewindShardLocationsRefresh(ecVolume *erasure_coding.EcVolume, d time.Durat
 func TestCachedLookupEcShardLocationsRefreshesAfterAFailedRead(t *testing.T) {
 	master, masterAddr := startCountingMaster(t)
 	store := &Store{
-		MasterAddress:  masterAddr,
+		MasterFn:       func() pb.ServerAddress { return masterAddr },
 		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
 	}
 	ecVolume := &erasure_coding.EcVolume{
@@ -127,7 +127,7 @@ func TestCachedLookupEcShardLocationsRefreshesAfterAFailedRead(t *testing.T) {
 func TestCachedLookupEcShardLocationsSettlesAfterRefresh(t *testing.T) {
 	master, masterAddr := startCountingMaster(t)
 	store := &Store{
-		MasterAddress:  masterAddr,
+		MasterFn:       func() pb.ServerAddress { return masterAddr },
 		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
 	}
 	ecVolume := &erasure_coding.EcVolume{
@@ -257,7 +257,7 @@ func TestReadOneEcShardIntervalMarksTheMapAfterADirectReadFails(t *testing.T) {
 func TestCachedLookupEcShardLocationsKeepsAMarkRaisedDuringTheLookup(t *testing.T) {
 	master, masterAddr := startCountingMaster(t)
 	store := &Store{
-		MasterAddress:  masterAddr,
+		MasterFn:       func() pb.ServerAddress { return masterAddr },
 		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
 	}
 	ecVolume := &erasure_coding.EcVolume{
@@ -295,7 +295,7 @@ func TestCachedLookupEcShardLocationsKeepsAMarkRaisedDuringTheLookup(t *testing.
 func TestCachedLookupEcShardLocationsKeepsTheMarkWhenTheLookupFails(t *testing.T) {
 	master, masterAddr := startCountingMaster(t)
 	store := &Store{
-		MasterAddress:  masterAddr,
+		MasterFn:       func() pb.ServerAddress { return masterAddr },
 		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
 	}
 	ecVolume := &erasure_coding.EcVolume{

weed/storage/store_ec_scrub.go

@@ -23,7 +23,7 @@ func (s *Store) ScrubEcVolume(vid needle.VolumeId, mode volume_server_pb.VolumeS
 		return 0, nil, []error{fmt.Errorf("EC volume id %d not found", vid)}
 	}
 	if err := s.cachedLookupEcShardLocations(ecv); err != nil {
-		return 0, nil, []error{fmt.Errorf("failed to locate shard via master grpc %s: %v", s.MasterAddress, err)}
+		return 0, nil, []error{fmt.Errorf("failed to locate shard via master grpc %s: %v", s.masterAddress(), err)}
 	}
 	return s.scrubEcVolumeWalk(ecv, mode, forceDeletedNeedlesCheck)
 }
@@ -56,7 +56,7 @@ func (s *Store) ScrubEcVolumeMerged(merged *erasure_coding.MergedEcRuntimes, mod
 		}
 	}
 	if err := s.cachedLookupEcShardLocations(ecv); err != nil {
-		return 0, nil, []error{fmt.Errorf("failed to locate shard via master grpc %s: %v", s.MasterAddress, err)}
+		return 0, nil, []error{fmt.Errorf("failed to locate shard via master grpc %s: %v", s.masterAddress(), err)}
 	}
 
 	files, shardInfos, errs := s.scrubEcVolumeWalk(ecv, mode, forceDeletedNeedlesCheck)