notification: assert delivery in the named targets test

The test only exercised SendMessage; it now records requests on both
endpoints and checks matching, non-matching, and unrouted keys reach
the right servers exactly once.

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)

seaweed-volume/src/storage/volume.rs

@@ -4185,8 +4185,10 @@ impl Volume {
         let _sync_guard = self.data_file_sync_lock.lock().unwrap();
         let _guard = self.data_file_access_control.write_lock();
         let (was_no_write_or_delete, was_no_write_can_delete) = self.no_write_flags();
-        // Remote-tiered volumes must stay no_write_can_delete regardless of marks.
-        let no_write_can_delete = was_no_write_can_delete && self.has_remote_file();
+        // A remote-tiered volume keeps the delete-only mode a load derives: its
+        // .dat cannot take appends, so "writable" resolves to can-delete rather
+        // than advertising writable with no local data file.
+        let no_write_can_delete = self.has_remote_file();
         self.no_write.store(
             if no_write_can_delete {
                 NO_WRITE_CAN_DELETE
@@ -4300,21 +4302,70 @@ impl Volume {
         }
         f(&mut self.volume_info.write().unwrap().files);
         let mut io = self.io.write().unwrap();
-        self.refresh_remote_write_mode(&mut io)
+        self.refresh_remote_write_mode(&mut io, true)
+    }
+
+    /// Go's MarkTieredDeletable: a live tier-up lifts the upload-time
+    /// read-only mark, except on a replicated volume — the shell copies the
+    /// uploaded replica's .idx and .vif to the other replicas after the
+    /// upload, and a tombstone written before that copy would reach one
+    /// replica only — or under an I/O quarantine, which is not the tier
+    /// upload's to lift. A refused volume keeps its flags until the next
+    /// load, as the Go volume does.
+    fn mark_tiered_deletable(&self) -> bool {
+        if self
+            .super_block
+            .read()
+            .unwrap()
+            .replica_placement
+            .has_replication()
+        {
+            return false;
+        }
+        if !self.has_remote_file() {
+            return false;
+        }
+        // Go takes noWriteLock around the quarantine check and the flag
+        // change; here a compare-exchange gives the same ordering — a
+        // heartbeat refusal landing between the check and the swap fails the
+        // swap, and the retry sees the quarantine mark the heartbeat set
+        // first.
+        loop {
+            if self.should_quarantine() {
+                return false;
+            }
+            let flags = self.no_write.load(Ordering::Acquire);
+            let promoted = (flags & !NO_WRITE_OR_DELETE) | NO_WRITE_CAN_DELETE;
+            if self
+                .no_write
+                .compare_exchange(flags, promoted, Ordering::AcqRel, Ordering::Acquire)
+                .is_ok()
+            {
+                return true;
+            }
+        }
     }
 
     /// Recompute the Go-style write/delete mode from the current remote tier
     /// state, and bring the needle map in line with it — a volume that stops
     /// being remote also stops using the read-only sorted index.
     ///
+    /// `live` is a tier transition on a running volume: the promotion then
+    /// goes through mark_tiered_deletable, Go's MarkTieredDeletable, while a
+    /// load promotes every tiered volume unconditionally as Go's load() does.
+    ///
     /// If the map cannot be rebuilt the volume is pinned read-only rather than
     /// published as writable with an index that rejects every put; a restart
     /// recovers it.
-    fn refresh_remote_write_mode(&self, io: &mut VolumeIo) -> Result<(), VolumeError> {
+    fn refresh_remote_write_mode(&self, io: &mut VolumeIo, live: bool) -> Result<(), VolumeError> {
         let has_remote = self.has_remote_file();
         if has_remote {
             // remote-tiered: no writes, deletes allowed
-            self.no_write.store(NO_WRITE_CAN_DELETE, Ordering::Release);
+            if !live {
+                self.no_write.store(NO_WRITE_CAN_DELETE, Ordering::Release);
+            } else {
+                self.mark_tiered_deletable();
+            }
         } else if !self.volume_info.read().unwrap().read_only_can_delete {
             // only clear the remoteness-derived flag, not an operator mark
             self.set_no_write_flag(NO_WRITE_CAN_DELETE, false);
@@ -4363,7 +4414,7 @@ impl Volume {
                     }
                 }
                 *self.volume_info.write().unwrap() = pb_info;
-                self.refresh_remote_write_mode(io)?;
+                self.refresh_remote_write_mode(io, false)?;
                 {
                     let mut vi = self.volume_info.write().unwrap();
                     if vi.version == 0 {
@@ -4394,7 +4445,7 @@ impl Volume {
                     self.set_no_write_flag(NO_WRITE_OR_DELETE, true);
                 }
                 *self.volume_info.write().unwrap() = pb_info;
-                self.refresh_remote_write_mode(io)?;
+                self.refresh_remote_write_mode(io, false)?;
                 {
                     let mut vi = self.volume_info.write().unwrap();
                     if vi.version == 0 {
@@ -9701,6 +9752,72 @@ mod tests {
         assert!(!v.is_no_write_or_delete());
     }
 
+    /// Go's MarkTieredDeletable leaves a replicated volume's upload-time
+    /// read-only mark alone: the shell still has to copy the uploaded
+    /// replica's .idx to the others, and a tombstone written before that copy
+    /// would reach one replica only.
+    #[test]
+    fn test_update_remote_files_keeps_a_replicated_volume_refusing_deletes() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let v = make_test_volume(dir);
+        v.super_block.write().unwrap().replica_placement =
+            crate::storage::super_block::ReplicaPlacement::from_string("001").unwrap();
+        v.set_no_write_or_delete(true); // the shell's pre-upload mark
+
+        v.update_remote_files(|files| files.push(Default::default()))
+            .unwrap();
+
+        assert!(v.has_remote_file());
+        assert!(
+            v.is_no_write_or_delete(),
+            "a replicated volume keeps refusing deletes until its next load"
+        );
+        assert!(!v.is_no_write_can_delete());
+    }
+
+    /// An I/O quarantine is not the tier upload's to lift: a live tier-up
+    /// leaves a quarantined volume's read-only mark alone, as Go's
+    /// MarkTieredDeletable does.
+    #[test]
+    fn test_update_remote_files_keeps_an_io_quarantine() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let v = make_test_volume(dir);
+        v.set_no_write_or_delete(true);
+        v.mark_io_quarantined();
+
+        v.update_remote_files(|files| files.push(Default::default()))
+            .unwrap();
+
+        assert!(v.has_remote_file());
+        assert!(v.is_no_write_or_delete());
+        assert!(!v.is_no_write_can_delete());
+    }
+
+    /// A refused tiered volume keeps OR_DELETE without CAN_DELETE; marking it
+    /// writable must resolve to the delete-only mode a load derives — the
+    /// remote .dat cannot take appends — not drop every protection.
+    #[test]
+    fn test_set_writable_on_a_refused_tiered_volume_stays_delete_only() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let v = make_test_volume(dir);
+        v.super_block.write().unwrap().replica_placement =
+            crate::storage::super_block::ReplicaPlacement::from_string("001").unwrap();
+        v.set_no_write_or_delete(true);
+
+        v.update_remote_files(|files| files.push(Default::default()))
+            .unwrap();
+        assert!(v.is_no_write_or_delete());
+
+        v.set_writable().unwrap();
+
+        assert!(v.is_read_only());
+        assert!(!v.no_write_flags().0);
+        assert!(v.no_write_flags().1);
+    }
+
     // Tier-down clears the remote mode and publishes the volume as writable. The
     // map it booted with is the read-only sorted one, whose put always fails, so
     // without a rebuild the first write appends to .dat and then cannot be

test/s3tables/table-buckets/verify_object_test.go

@@ -0,0 +1,139 @@
+package s3tables
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/smithy-go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func newObjectClient(t *testing.T) *s3.Client {
+	t.Helper()
+	cfg, err := config.LoadDefaultConfig(context.Background(),
+		config.WithRegion(testRegion),
+		config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(testAccessKey, testSecretKey, "")),
+	)
+	require.NoError(t, err)
+	return s3.NewFromConfig(cfg, func(o *s3.Options) {
+		o.BaseEndpoint = aws.String(sharedCluster.s3Endpoint)
+		o.UsePathStyle = true
+	})
+}
+
+func apiErrorCode(err error) string {
+	var apiErr smithy.APIError
+	if errors.As(err, &apiErr) {
+		return apiErr.ErrorCode()
+	}
+	return ""
+}
+
+// The external-volume verification object round-trips at a table bucket's
+// root the way the check uses it: PUT, HEAD, GET, list, DELETE. Any other key
+// at the root is still refused.
+func TestTableBucketExternalVolumeVerificationRoundTrip(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+	ctx := context.Background()
+	tables := NewS3TablesClient(sharedCluster.s3Endpoint, testRegion, testAccessKey, testSecretKey)
+	bucket := "verify-" + randomString(8)
+	created, err := tables.CreateTableBucket(bucket, nil)
+	require.NoError(t, err)
+	defer tables.DeleteTableBucket(created.ARN)
+	client := newObjectClient(t)
+
+	body := []byte("verification payload!") // 21 bytes
+	key := fmt.Sprintf("verify_%d_%d", time.Now().UnixMilli(), 1081846483)
+
+	// the control: an ordinary key at the root is refused
+	_, err = client.PutObject(ctx, &s3.PutObjectInput{Bucket: aws.String(bucket), Key: aws.String("notes.parquet"), Body: bytes.NewReader(body)})
+	require.Error(t, err)
+	assert.Equal(t, "AccessDenied", apiErrorCode(err))
+
+	_, err = client.PutObject(ctx, &s3.PutObjectInput{Bucket: aws.String(bucket), Key: aws.String(key), Body: bytes.NewReader(body), ContentLength: aws.Int64(int64(len(body)))})
+	require.NoError(t, err)
+
+	head, err := client.HeadObject(ctx, &s3.HeadObjectInput{Bucket: aws.String(bucket), Key: aws.String(key)})
+	require.NoError(t, err)
+	assert.Equal(t, int64(len(body)), aws.ToInt64(head.ContentLength))
+
+	got, err := client.GetObject(ctx, &s3.GetObjectInput{Bucket: aws.String(bucket), Key: aws.String(key)})
+	require.NoError(t, err)
+	data, err := io.ReadAll(got.Body)
+	got.Body.Close()
+	require.NoError(t, err)
+	assert.Equal(t, body, data)
+
+	list, err := client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{Bucket: aws.String(bucket), Prefix: aws.String("verify_")})
+	require.NoError(t, err)
+	var keys []string
+	for _, o := range list.Contents {
+		keys = append(keys, aws.ToString(o.Key))
+	}
+	assert.Equal(t, []string{key}, keys)
+
+	_, err = client.DeleteObject(ctx, &s3.DeleteObjectInput{Bucket: aws.String(bucket), Key: aws.String(key)})
+	require.NoError(t, err)
+	_, err = client.HeadObject(ctx, &s3.HeadObjectInput{Bucket: aws.String(bucket), Key: aws.String(key)})
+	require.Error(t, err, "the verification object must be gone after DELETE")
+}
+
+// A namespace can carry a name of the verification key's shape. The key
+// then names the namespace's directory: a PUT or DELETE through it must not
+// replace or remove the namespace.
+func TestTableBucketVerificationKeyLeavesNamespaceAlone(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+	ctx := context.Background()
+	tables := NewS3TablesClient(sharedCluster.s3Endpoint, testRegion, testAccessKey, testSecretKey)
+	bucket := "verify-ns-" + randomString(8)
+	created, err := tables.CreateTableBucket(bucket, nil)
+	require.NoError(t, err)
+	defer tables.DeleteTableBucket(created.ARN)
+	client := newObjectClient(t)
+	body := []byte("verification payload!")
+
+	listed := func(name string) bool {
+		resp, err := tables.ListNamespaces(created.ARN, "", "", 0)
+		require.NoError(t, err)
+		for _, ns := range resp.Namespaces {
+			if len(ns.Namespace) == 1 && ns.Namespace[0] == name {
+				return true
+			}
+		}
+		return false
+	}
+	for i, op := range []string{"PUT", "DELETE"} {
+		t.Run(op, func(t *testing.T) {
+			name := fmt.Sprintf("verify_%d_%d", time.Now().UnixMilli(), 7+i)
+			_, err := tables.CreateNamespace(created.ARN, []string{name})
+			require.NoError(t, err)
+			defer tables.DeleteNamespace(created.ARN, []string{name})
+			require.True(t, listed(name), "setup: the namespace is listed")
+
+			if op == "PUT" {
+				_, err = client.PutObject(ctx, &s3.PutObjectInput{Bucket: aws.String(bucket), Key: aws.String(name), Body: bytes.NewReader(body), ContentLength: aws.Int64(int64(len(body)))})
+				assert.Error(t, err, "a PUT over a namespace must be refused")
+			} else {
+				_, err = client.DeleteObject(ctx, &s3.DeleteObjectInput{Bucket: aws.String(bucket), Key: aws.String(name)})
+				assert.Error(t, err, "a DELETE of a namespace must be refused")
+			}
+			_, err = tables.GetNamespace(created.ARN, []string{name})
+			assert.NoError(t, err, "the namespace survives the %s", op)
+			assert.True(t, listed(name), "the namespace is still listed after the %s", op)
+		})
+	}
+}

weed/command/scaffold/master.toml

@@ -28,6 +28,10 @@ region = "us-east-2"
 bucket = "your_bucket_name"    # an existing bucket
 endpoint = ""
 storage_class = "STANDARD_IA"
+# optional HTTP client tuning for S3-compatible services:
+# disable_http2 = false                 # true keeps the client on HTTP/1.1, one connection per concurrent request
+# max_connections = 0                   # cap on connections to the endpoint, 0 = no cap
+# idle_connection_timeout_seconds = 0   # close idle connections sooner, 0 = Go default (90s)
 
 # create this number of logical volumes if no more writable volumes
 # count_x means how many copies of data.

weed/command/scaffold/notification.toml

@@ -81,3 +81,16 @@ workers = 5                                        # optional: concurrent worker
 buffer_size = 10000                                # optional: event buffer size (default: 10000, range: 100-1000000)
 # event_types = ["create", "update", "delete", "rename"]  # optional: filter by event types (default: all)
 # path_prefixes = ["/important", "/data"]          # optional: filter by path prefixes (default: all)
+#
+# Additional endpoints can be declared as named subsections, each with its own
+# endpoint, filters, retry and worker settings, e.g.:
+# [notification.webhook.file_ingress]
+# enabled = true
+# endpoint = "https://ingress.example.com/hook"
+# event_types = ["create"]
+# path_prefixes = ["/buckets/uploads/data"]
+#
+# [notification.webhook.metadata_extractor]
+# enabled = true
+# endpoint = "https://metadata.example.com/hook"
+# path_prefixes = ["/buckets/repository"]

weed/notification/configuration.go

@@ -26,10 +26,10 @@ func LoadConfiguration(config *util.ViperProxy, prefix string) {
 		return
 	}
 
-	validateOneEnabledQueue(config)
+	validateOneEnabledQueue(config, prefix)
 
 	for _, queue := range MessageQueues {
-		if config.GetBool(prefix + queue.GetName() + ".enabled") {
+		if queueEnabled(config, prefix, queue) {
 			if err := queue.Initialize(config, prefix+queue.GetName()+"."); err != nil {
 				glog.Fatalf("Failed to initialize notification for %s: %+v",
 					queue.GetName(), err)
@@ -42,10 +42,26 @@ func LoadConfiguration(config *util.ViperProxy, prefix string) {
 
 }
 
-func validateOneEnabledQueue(config *util.ViperProxy) {
+// queueEnabled reports whether a queue is switched on — by the flat
+// "<prefix><name>.enabled" flag or, for queues accepting named sub-instances
+// ([notification.<queue>.<target>]), by any enabled subconfig.
+func queueEnabled(config util.Configuration, prefix string, queue MessageQueue) bool {
+	key := prefix + queue.GetName() + "."
+	if config.GetBool(key + "enabled") {
+		return true
+	}
+	if q, ok := queue.(interface {
+		HasEnabledSubconfig(util.Configuration, string) bool
+	}); ok {
+		return q.HasEnabledSubconfig(config, key)
+	}
+	return false
+}
+
+func validateOneEnabledQueue(config *util.ViperProxy, prefix string) {
 	enabledQueue := ""
 	for _, queue := range MessageQueues {
-		if config.GetBool(queue.GetName() + ".enabled") {
+		if queueEnabled(config, prefix, queue) {
 			if enabledQueue == "" {
 				enabledQueue = queue.GetName()
 			} else {

weed/notification/webhook/types.go

@@ -3,7 +3,6 @@ package webhook
 import (
 	"fmt"
 	"net/url"
-	"strconv"
 
 	"github.com/seaweedfs/seaweedfs/weed/notification"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -30,12 +29,6 @@ func (e eventType) valid() bool {
 	return notification.ValidEventType(string(e))
 }
 
-var (
-	pubSubHandlerNameTemplate = func(n int) string {
-		return "webhook_handler_" + strconv.Itoa(n)
-	}
-)
-
 type client interface {
 	sendMessage(message *webhookMessage) error
 }

weed/notification/webhook/webhook_queue.go

@@ -4,6 +4,9 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sort"
+	"strings"
+	"sync"
 	"time"
 
 	"github.com/ThreeDotsLabs/watermill"
@@ -22,45 +25,145 @@ func init() {
 	notification.MessageQueues = append(notification.MessageQueues, &Queue{})
 }
 
+// target is one webhook endpoint with its own filter, http client and
+// concurrency budget. The flat [notification.webhook] section is the unnamed
+// target; each enabled [notification.webhook.<name>] subsection adds another.
+type target struct {
+	name   string
+	config *config
+	filter *filter
+	client client
+	sem    chan struct{}
+	inbox  chan *message.Message
+}
+
+func newTarget(name string, cfg *config) (*target, error) {
+	t := &target{
+		name:   name,
+		config: cfg,
+		filter: newFilter(cfg),
+		sem:    make(chan struct{}, cfg.nWorkers),
+		inbox:  make(chan *message.Message, cfg.bufferSize),
+	}
+	hClient, err := newHTTPClient(cfg)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create webhook http client: %w", err)
+	}
+	t.client = hClient
+	return t, nil
+}
+
+func (t *target) topic() string {
+	if t.name == "" {
+		return pubSubTopicName
+	}
+	return pubSubTopicName + "_" + t.name
+}
+
+func (t *target) handlerName() string {
+	if t.name == "" {
+		return "webhook_handler"
+	}
+	return "webhook_handler_" + t.name
+}
+
 type Queue struct {
-	router       *message.Router
+	routers      []*message.Router
 	queueChannel *gochannel.GoChannel
-	config       *config
-	client       client
-	filter       *filter
+	targets      []*target
 
 	ctx    context.Context
 	cancel context.CancelFunc
-
-	// Semaphore for controlling concurrent webhook requests
-	sem chan struct{}
 }
 
 func (w *Queue) GetName() string {
 	return queueName
 }
 
+// HasEnabledSubconfig reports whether any [notification.webhook.<name>]
+// subsection is enabled, so the queue activates on named targets even when the
+// flat [notification.webhook] section is absent.
+func (w *Queue) HasEnabledSubconfig(configuration util.Configuration, prefix string) bool {
+	return len(namedTargetPrefixes(configuration, prefix)) > 0
+}
+
+// stringMapConfig is implemented by *util.ViperProxy; anything narrower simply
+// gets no named-target discovery.
+type stringMapConfig interface {
+	GetStringMap(key string) map[string]interface{}
+}
+
+func namedTargetPrefixes(configuration util.Configuration, prefix string) []string {
+	mg, ok := configuration.(stringMapConfig)
+	if !ok {
+		return nil
+	}
+	var prefixes []string
+	for k, v := range mg.GetStringMap(strings.TrimSuffix(prefix, ".")) {
+		if _, ok := v.(map[string]interface{}); !ok {
+			continue
+		}
+		p := prefix + k + "."
+		if !configuration.GetBool(p + "enabled") {
+			continue
+		}
+		prefixes = append(prefixes, p)
+	}
+	sort.Strings(prefixes)
+	return prefixes
+}
+
+func buildTargets(configuration util.Configuration, prefix string) ([]*target, error) {
+	var prefixes []string
+	if configuration.GetBool(prefix + "enabled") {
+		prefixes = append(prefixes, prefix)
+	}
+	prefixes = append(prefixes, namedTargetPrefixes(configuration, prefix)...)
+
+	var targets []*target
+	for _, p := range prefixes {
+		cfg := newConfigWithDefaults(configuration, p)
+		if err := cfg.validate(); err != nil {
+			return nil, err
+		}
+		t, err := newTarget(strings.TrimSuffix(strings.TrimPrefix(p, prefix), "."), cfg)
+		if err != nil {
+			return nil, err
+		}
+		targets = append(targets, t)
+	}
+	return targets, nil
+}
+
 func (w *Queue) SendMessage(key string, msg proto.Message) error {
 	eventNotification, ok := msg.(*filer_pb.EventNotification)
 	if !ok {
 		return nil
 	}
 
-	if w.filter != nil && !w.filter.shouldPublish(key, eventNotification) {
-		return nil
+	var firstErr error
+	for _, t := range w.targets {
+		if t.filter != nil && !t.filter.shouldPublish(key, eventNotification) {
+			continue
+		}
+		m := newWebhookMessage(key, msg)
+		if m == nil {
+			continue
+		}
+		wMsg, err := m.toWaterMillMessage()
+		if err != nil {
+			return err
+		}
+		wMsg.Metadata.Set("target", t.name)
+		select {
+		case t.inbox <- wMsg:
+		default:
+			if firstErr == nil {
+				firstErr = fmt.Errorf("webhook target %q buffer is full, dropping event for %s", t.name, key)
+			}
+		}
 	}
-
-	m := newWebhookMessage(key, msg)
-	if m == nil {
-		return nil
-	}
-
-	wMsg, err := m.toWaterMillMessage()
-	if err != nil {
-		return err
-	}
-
-	return w.queueChannel.Publish(pubSubTopicName, wMsg)
+	return firstErr
 }
 
 func (w *webhookMessage) toWaterMillMessage() (*message.Message, error) {
@@ -78,105 +181,128 @@ func (w *webhookMessage) toWaterMillMessage() (*message.Message, error) {
 }
 
 func (w *Queue) Initialize(configuration util.Configuration, prefix string) error {
-	c := newConfigWithDefaults(configuration, prefix)
-
-	if err := c.validate(); err != nil {
+	targets, err := buildTargets(configuration, prefix)
+	if err != nil {
 		return err
 	}
+	if len(targets) == 0 {
+		return fmt.Errorf("webhook endpoint is required")
+	}
 
-	return w.initialize(c)
+	return w.initialize(targets)
 }
 
-func (w *Queue) initialize(cfg *config) error {
+func (w *Queue) initialize(targets []*target) error {
 	w.ctx, w.cancel = context.WithCancel(context.Background())
-	w.config = cfg
-	w.filter = newFilter(cfg)
+	w.targets = targets
 
-	// Initialize semaphore for controlling concurrent webhook requests
-	w.sem = make(chan struct{}, cfg.nWorkers)
-
-	hClient, err := newHTTPClient(cfg)
-	if err != nil {
-		return fmt.Errorf("failed to create webhook http client: %w", err)
-	}
-	w.client = hClient
-
-	if err = w.setupWatermillQueue(cfg); err != nil {
+	if err := w.setupWatermillQueue(targets); err != nil {
 		return fmt.Errorf("failed to setup watermill queue: %w", err)
 	}
-	if err = w.logDeadLetterMessages(); err != nil {
+	if err := w.logDeadLetterMessages(); err != nil {
 		return err
 	}
 
 	return nil
 }
 
-func (w *Queue) setupWatermillQueue(cfg *config) error {
+func (w *Queue) setupWatermillQueue(targets []*target) error {
 	logger := watermill.NewStdLogger(false, false)
+	bufferSize := 0
+	for _, t := range targets {
+		if t.config.bufferSize > bufferSize {
+			bufferSize = t.config.bufferSize
+		}
+	}
 	pubSubConfig := gochannel.Config{
-		OutputChannelBuffer: int64(cfg.bufferSize),
+		OutputChannelBuffer: int64(bufferSize),
 		Persistent:          false,
 	}
 	w.queueChannel = gochannel.NewGoChannel(pubSubConfig, logger)
 
-	router, err := message.NewRouter(
-		message.RouterConfig{
-			CloseTimeout: 60 * time.Second,
-		},
-		logger,
-	)
-	if err != nil {
-		return fmt.Errorf("failed to create router: %w", err)
-	}
-	w.router = router
+	var wg sync.WaitGroup
+	for _, t := range targets {
+		// One router per target so each keeps its own retry/backoff settings.
+		// Handlers subscribe to the target's own topic, so a message is
+		// delivered to an endpoint exactly once per matching target.
+		router, err := message.NewRouter(
+			message.RouterConfig{
+				CloseTimeout: 60 * time.Second,
+			},
+			logger,
+		)
+		if err != nil {
+			return fmt.Errorf("failed to create router: %w", err)
+		}
+		w.routers = append(w.routers, router)
 
-	retryMiddleware := middleware.Retry{
-		MaxRetries:          cfg.maxRetries,
-		InitialInterval:     time.Duration(cfg.backoffSeconds) * time.Second,
-		MaxInterval:         time.Duration(cfg.maxBackoffSeconds) * time.Second,
-		Multiplier:          2.0,
-		RandomizationFactor: 0.3,
-		Logger:              logger,
-	}.Middleware
+		retryMiddleware := middleware.Retry{
+			MaxRetries:          t.config.maxRetries,
+			InitialInterval:     time.Duration(t.config.backoffSeconds) * time.Second,
+			MaxInterval:         time.Duration(t.config.maxBackoffSeconds) * time.Second,
+			Multiplier:          2.0,
+			RandomizationFactor: 0.3,
+			Logger:              logger,
+		}.Middleware
 
-	poisonQueue, err := middleware.PoisonQueue(w.queueChannel, deadLetterTopic)
-	if err != nil {
-		return fmt.Errorf("failed to create poison queue: %w", err)
-	}
-
-	router.AddPlugin(plugin.SignalsHandler)
-	router.AddMiddleware(retryMiddleware, poisonQueue)
-
-	// Add a single handler to avoid duplicate message delivery.
-	// With gochannel's default behavior, each handler call creates
-	// a separate subscription, and all subscriptions receive their own copy of each message.
-	// Using a single handler ensures each webhook is sent only once.
-	// Concurrency is controlled via semaphore in handleWebhook based on nWorkers config.
-	router.AddConsumerHandler(
-		"webhook_handler",
-		pubSubTopicName,
-		w.queueChannel,
-		w.handleWebhook,
-	)
-
-	go func() {
-		// cancels the queue context so the dead letter logger exists in case context not canceled by the shutdown signal already
-		defer w.cancel()
-
-		if err := router.Run(w.ctx); err != nil && !errors.Is(err, context.Canceled) {
-			glog.Errorf("webhook pubsub worker stopped with error: %v", err)
+		poisonQueue, err := middleware.PoisonQueue(w.queueChannel, deadLetterTopic)
+		if err != nil {
+			return fmt.Errorf("failed to create poison queue: %w", err)
 		}
 
-		glog.Info("webhook pubsub worker stopped")
+		router.AddPlugin(plugin.SignalsHandler)
+		router.AddMiddleware(retryMiddleware, poisonQueue)
+
+		t := t
+		router.AddConsumerHandler(
+			t.handlerName(),
+			t.topic(),
+			w.queueChannel,
+			t.handleWebhook,
+		)
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if err := router.Run(w.ctx); err != nil && !errors.Is(err, context.Canceled) {
+				glog.Errorf("webhook pubsub worker stopped with error: %v", err)
+			}
+			glog.Info("webhook pubsub worker stopped")
+		}()
+
+		go w.forwardTargetMessages(t)
+	}
+
+	// the queue context lives until every target's worker has stopped so the
+	// dead letter logger keeps running while any of them is still up
+	go func() {
+		wg.Wait()
+		w.cancel()
 	}()
 
 	return nil
 }
 
-func (w *Queue) handleWebhook(msg *message.Message) error {
+// forwardTargetMessages publishes each queued event to the target's topic in
+// order. It owns the potentially blocking Publish call so a saturated target
+// can neither stall the other targets nor the caller of SendMessage.
+func (w *Queue) forwardTargetMessages(t *target) {
+	for {
+		select {
+		case msg := <-t.inbox:
+			if err := w.queueChannel.Publish(t.topic(), msg); err != nil {
+				glog.Errorf("webhook target %q failed to publish message: %v", t.name, err)
+			}
+		case <-w.ctx.Done():
+			return
+		}
+	}
+}
+
+func (t *target) handleWebhook(msg *message.Message) error {
 	// Acquire semaphore slot (blocks if at capacity)
-	w.sem <- struct{}{}
-	defer func() { <-w.sem }()
+	t.sem <- struct{}{}
+	defer func() { <-t.sem }()
 
 	var n filer_pb.EventNotification
 	if err := proto.Unmarshal(msg.Payload, &n); err != nil {
@@ -191,8 +317,8 @@ func (w *Queue) handleWebhook(msg *message.Message) error {
 		Notification: &n,
 	}
 
-	if err := w.client.sendMessage(webhookMsg); err != nil {
-		glog.Errorf("failed to send message to webhook %s: %v", webhookMsg.Key, err)
+	if err := t.client.sendMessage(webhookMsg); err != nil {
+		glog.Errorf("webhook target %q failed to send message for %s: %v", t.name, webhookMsg.Key, err)
 		return err
 	}
 
@@ -218,10 +344,12 @@ func (w *Queue) logDeadLetterMessages() error {
 					continue
 				}
 				key := "unknown"
+				targetName := ""
 				if msg.Metadata != nil {
 					if keyValue, exists := msg.Metadata["key"]; exists {
 						key = keyValue
 					}
+					targetName = msg.Metadata["target"]
 				}
 				payload := ""
 				if msg.Payload != nil {
@@ -232,7 +360,7 @@ func (w *Queue) logDeadLetterMessages() error {
 						payload = n.String()
 					}
 				}
-				glog.Errorf("received dead letter message: %s, key: %s", payload, key)
+				glog.Errorf("received dead letter message: %s, key: %s, target: %s", payload, key, targetName)
 			case <-w.ctx.Done():
 				return
 			}

weed/notification/webhook/webhook_queue_test.go

@@ -4,6 +4,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -170,6 +171,29 @@ func TestWebhookMessageSerialization(t *testing.T) {
 	}
 }
 
+func newQueueForTest(t *testing.T, cfg *config) *Queue {
+	t.Helper()
+	tgt, err := newTarget("", cfg)
+	if err != nil {
+		t.Fatalf("newTarget() error = %v", err)
+	}
+	q := &Queue{}
+	if err := q.initialize([]*target{tgt}); err != nil {
+		t.Fatalf("initialize() error = %v", err)
+	}
+	return q
+}
+
+func closeQueueForTest(q *Queue) {
+	if q.cancel != nil {
+		q.cancel()
+	}
+	time.Sleep(100 * time.Millisecond)
+	for _, r := range q.routers {
+		r.Close()
+	}
+}
+
 func TestQueueInitialize(t *testing.T) {
 	cfg := &config{
 		endpoint:          "https://example.com/webhook",
@@ -182,32 +206,19 @@ func TestQueueInitialize(t *testing.T) {
 		bufferSize:        100,
 	}
 
-	q := &Queue{}
-	err := q.initialize(cfg)
-	if err != nil {
-		t.Errorf("Initialize() error = %v", err)
-	}
+	q := newQueueForTest(t, cfg)
+	defer closeQueueForTest(q)
 
-	defer func() {
-		if q.cancel != nil {
-			q.cancel()
-		}
-		time.Sleep(100 * time.Millisecond)
-		if q.router != nil {
-			q.router.Close()
-		}
-	}()
-
-	if q.router == nil {
+	if len(q.routers) != 1 {
 		t.Error("Expected router to be initialized")
 	}
 	if q.queueChannel == nil {
 		t.Error("Expected queueChannel to be initialized")
 	}
-	if q.client == nil {
+	if q.targets[0].client == nil {
 		t.Error("Expected client to be initialized")
 	}
-	if q.config == nil {
+	if q.targets[0].config == nil {
 		t.Error("Expected config to be initialized")
 	}
 }
@@ -225,21 +236,8 @@ func TestQueueSendMessage(t *testing.T) {
 		bufferSize:        10,
 	}
 
-	q := &Queue{}
-	err := q.initialize(cfg)
-	if err != nil {
-		t.Fatalf("Failed to initialize queue: %v", err)
-	}
-
-	defer func() {
-		if q.cancel != nil {
-			q.cancel()
-		}
-		time.Sleep(100 * time.Millisecond)
-		if q.router != nil {
-			q.router.Close()
-		}
-	}()
+	q := newQueueForTest(t, cfg)
+	defer closeQueueForTest(q)
 
 	msg := &filer_pb.EventNotification{
 		NewEntry: &filer_pb.Entry{
@@ -247,8 +245,7 @@ func TestQueueSendMessage(t *testing.T) {
 		},
 	}
 
-	err = q.SendMessage("/test/path", msg)
-	if err != nil {
+	if err := q.SendMessage("/test/path", msg); err != nil {
 		t.Errorf("SendMessage() error = %v", err)
 	}
 }
@@ -271,7 +268,7 @@ func TestQueueHandleWebhook(t *testing.T) {
 	}
 
 	client, _ := newHTTPClient(cfg)
-	q := &Queue{
+	tgt := &target{
 		client: client,
 		sem:    make(chan struct{}, cfg.nWorkers),
 	}
@@ -287,8 +284,7 @@ func TestQueueHandleWebhook(t *testing.T) {
 		t.Fatalf("Failed to create watermill message: %v", err)
 	}
 
-	err = q.handleWebhook(wmMsg)
-	if err != nil {
+	if err := tgt.handleWebhook(wmMsg); err != nil {
 		t.Errorf("handleWebhook() error = %v", err)
 	}
 }
@@ -307,21 +303,8 @@ func TestQueueEndToEnd(t *testing.T) {
 		bufferSize:        10,
 	}
 
-	q := &Queue{}
-	err := q.initialize(cfg)
-	if err != nil {
-		t.Fatalf("Failed to initialize queue: %v", err)
-	}
-
-	defer func() {
-		if q.cancel != nil {
-			q.cancel()
-		}
-		time.Sleep(100 * time.Millisecond)
-		if q.router != nil {
-			q.router.Close()
-		}
-	}()
+	q := newQueueForTest(t, cfg)
+	defer closeQueueForTest(q)
 
 	msg := &filer_pb.EventNotification{
 		NewEntry: &filer_pb.Entry{
@@ -329,8 +312,7 @@ func TestQueueEndToEnd(t *testing.T) {
 		},
 	}
 
-	err = q.SendMessage("/test/path", msg)
-	if err != nil {
+	if err := q.SendMessage("/test/path", msg); err != nil {
 		t.Errorf("SendMessage() error = %v", err)
 	}
 }
@@ -347,33 +329,20 @@ func TestQueueRetryMechanism(t *testing.T) {
 		bufferSize:        10,
 	}
 
-	q := &Queue{}
-	err := q.initialize(cfg)
-	if err != nil {
-		t.Fatalf("Failed to initialize queue: %v", err)
-	}
-
-	defer func() {
-		if q.cancel != nil {
-			q.cancel()
-		}
-		time.Sleep(100 * time.Millisecond)
-		if q.router != nil {
-			q.router.Close()
-		}
-	}()
+	q := newQueueForTest(t, cfg)
+	defer closeQueueForTest(q)
 
 	// Verify that the queue is properly configured for retries
-	if q.config.maxRetries != 3 {
-		t.Errorf("Expected maxRetries=3, got %d", q.config.maxRetries)
+	if q.targets[0].config.maxRetries != 3 {
+		t.Errorf("Expected maxRetries=3, got %d", q.targets[0].config.maxRetries)
 	}
 
-	if q.config.backoffSeconds != 2 {
-		t.Errorf("Expected backoffSeconds=2, got %d", q.config.backoffSeconds)
+	if q.targets[0].config.backoffSeconds != 2 {
+		t.Errorf("Expected backoffSeconds=2, got %d", q.targets[0].config.backoffSeconds)
 	}
 
-	if q.config.maxBackoffSeconds != 10 {
-		t.Errorf("Expected maxBackoffSeconds=10, got %d", q.config.maxBackoffSeconds)
+	if q.targets[0].config.maxBackoffSeconds != 10 {
+		t.Errorf("Expected maxBackoffSeconds=10, got %d", q.targets[0].config.maxBackoffSeconds)
 	}
 
 	// Test that we can send a message (retry behavior is handled by Watermill middleware)
@@ -381,8 +350,7 @@ func TestQueueRetryMechanism(t *testing.T) {
 		NewEntry: &filer_pb.Entry{Name: "test.txt"},
 	}
 
-	err = q.SendMessage("/test/retry", msg)
-	if err != nil {
+	if err := q.SendMessage("/test/retry", msg); err != nil {
 		t.Errorf("SendMessage() error = %v", err)
 	}
 }
@@ -419,21 +387,8 @@ func TestQueueNoDuplicateWebhooks(t *testing.T) {
 				bufferSize:        10,
 			}
 
-			q := &Queue{}
-			err := q.initialize(cfg)
-			if err != nil {
-				t.Fatalf("Failed to initialize queue: %v", err)
-			}
-
-			defer func() {
-				if q.cancel != nil {
-					q.cancel()
-				}
-				time.Sleep(100 * time.Millisecond)
-				if q.router != nil {
-					q.router.Close()
-				}
-			}()
+			q := newQueueForTest(t, cfg)
+			defer closeQueueForTest(q)
 
 			// Wait for router and subscriber to be fully ready
 			time.Sleep(200 * time.Millisecond)
@@ -444,8 +399,7 @@ func TestQueueNoDuplicateWebhooks(t *testing.T) {
 				},
 			}
 
-			err = q.SendMessage("/test/path", msg)
-			if err != nil {
+			if err := q.SendMessage("/test/path", msg); err != nil {
 				t.Errorf("SendMessage() error = %v", err)
 			}
 
@@ -607,3 +561,146 @@ func TestQueueSendMessageWithFilter(t *testing.T) {
 		})
 	}
 }
+
+type mapConfig struct {
+	values map[string]interface{}
+}
+
+func (m *mapConfig) GetString(key string) string {
+	v, _ := m.values[key].(string)
+	return v
+}
+
+func (m *mapConfig) GetBool(key string) bool {
+	v, _ := m.values[key].(bool)
+	return v
+}
+
+func (m *mapConfig) GetInt(key string) int {
+	v, _ := m.values[key].(int)
+	return v
+}
+
+func (m *mapConfig) GetStringSlice(key string) []string {
+	v, _ := m.values[key].([]string)
+	return v
+}
+
+func (m *mapConfig) SetDefault(key string, value interface{}) {}
+
+func (m *mapConfig) GetStringMap(key string) map[string]interface{} {
+	v, _ := m.values[key].(map[string]interface{})
+	return v
+}
+
+func TestNamedTargets(t *testing.T) {
+	var hitsA, hitsB atomic.Int32
+	serverA := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		hitsA.Add(1)
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer serverA.Close()
+	serverB := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		hitsB.Add(1)
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer serverB.Close()
+
+	cfg := &mapConfig{values: map[string]interface{}{
+		"notification.webhook": map[string]interface{}{
+			"file_ingress": map[string]interface{}{
+				"enabled":       true,
+				"endpoint":      serverA.URL,
+				"path_prefixes": []string{"/data/"},
+			},
+			"metadata_extractor": map[string]interface{}{
+				"enabled":       true,
+				"endpoint":      serverB.URL,
+				"path_prefixes": []string{"/repository/"},
+			},
+			"disabled_one": map[string]interface{}{
+				"enabled":  false,
+				"endpoint": "https://example.com/never",
+			},
+		},
+		"notification.webhook.file_ingress.enabled":             true,
+		"notification.webhook.file_ingress.endpoint":            serverA.URL,
+		"notification.webhook.file_ingress.path_prefixes":       []string{"/data/"},
+		"notification.webhook.metadata_extractor.enabled":       true,
+		"notification.webhook.metadata_extractor.endpoint":      serverB.URL,
+		"notification.webhook.metadata_extractor.path_prefixes": []string{"/repository/"},
+	}}
+
+	q := &Queue{}
+	if !q.HasEnabledSubconfig(cfg, "notification.webhook.") {
+		t.Fatal("expected named subconfig to be detected as enabled")
+	}
+
+	targets, err := buildTargets(cfg, "notification.webhook.")
+	if err != nil {
+		t.Fatalf("buildTargets() error = %v", err)
+	}
+	if len(targets) != 2 {
+		t.Fatalf("expected 2 targets, got %d", len(targets))
+	}
+	if targets[0].name != "file_ingress" || targets[1].name != "metadata_extractor" {
+		t.Fatalf("unexpected target names: %s, %s", targets[0].name, targets[1].name)
+	}
+
+	if err := q.initialize(targets); err != nil {
+		t.Fatalf("initialize() error = %v", err)
+	}
+	defer closeQueueForTest(q)
+
+	time.Sleep(200 * time.Millisecond)
+
+	msg := &filer_pb.EventNotification{
+		NewEntry: &filer_pb.Entry{Name: "f.txt"},
+	}
+	for _, key := range []string{"/data/f.txt", "/repository/f.txt", "/elsewhere/f.txt"} {
+		if err := q.SendMessage(key, msg); err != nil {
+			t.Errorf("SendMessage(%q) error = %v", key, err)
+		}
+	}
+
+	deadline := time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) && (hitsA.Load() == 0 || hitsB.Load() == 0) {
+		time.Sleep(50 * time.Millisecond)
+	}
+	time.Sleep(200 * time.Millisecond)
+
+	if got := hitsA.Load(); got != 1 {
+		t.Errorf("expected file_ingress to receive 1 event, got %d", got)
+	}
+	if got := hitsB.Load(); got != 1 {
+		t.Errorf("expected metadata_extractor to receive 1 event, got %d", got)
+	}
+}
+
+// A disabled flat section keeps no target even when an endpoint is set —
+// the scaffold ships exactly that combination, and a named target must not
+// reactivate it.
+func TestBuildTargetsDisabledFlat(t *testing.T) {
+	cfg := &mapConfig{values: map[string]interface{}{
+		"notification.webhook": map[string]interface{}{
+			"enabled":  false,
+			"endpoint": "https://your-server.com/webhook",
+			"named": map[string]interface{}{
+				"enabled":  true,
+				"endpoint": "https://example.com/hook",
+			},
+		},
+		"notification.webhook.enabled":        false,
+		"notification.webhook.endpoint":       "https://your-server.com/webhook",
+		"notification.webhook.named.enabled":  true,
+		"notification.webhook.named.endpoint": "https://example.com/hook",
+	}}
+
+	targets, err := buildTargets(cfg, "notification.webhook.")
+	if err != nil {
+		t.Fatalf("buildTargets() error = %v", err)
+	}
+	if len(targets) != 1 || targets[0].name != "named" {
+		t.Fatalf("expected only the named target, got %v", targets)
+	}
+}

weed/s3api/auth_credentials.go

@@ -25,6 +25,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3bucket"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
 	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
 	"github.com/seaweedfs/seaweedfs/weed/wdclient"
@@ -1680,6 +1681,7 @@ func (iam *IdentityAccessManagement) Auth(f http.HandlerFunc, action Action) htt
 		// succeeds but authZ denies, so the denied request still audits its
 		// requester; a failed authN resolves no identity at all.
 		identity, errCode, _ := iam.authRequestWithAuthType(r, action)
+		errCode = refineDenialForBucketName(r, identity, errCode)
 		if errCode != s3err.ErrNone {
 			glog.V(3).Infof("auth error: %v", errCode)
 		}
@@ -1719,6 +1721,7 @@ func (iam *IdentityAccessManagement) AuthPostPolicy(f http.HandlerFunc, action A
 			}
 		}
 
+		errCode = refineDenialForBucketName(r, identity, errCode)
 		if errCode != s3err.ErrNone {
 			glog.V(3).Infof("auth error: %v", errCode)
 		}
@@ -1727,6 +1730,39 @@ func (iam *IdentityAccessManagement) AuthPostPolicy(f http.HandlerFunc, action A
 	}
 }
 
+// refineDenialForBucketName answers 400 InvalidBucketName, as AWS S3 does,
+// when an authenticated caller is denied on a bucket whose name breaks the S3
+// naming rules. The rules are public and the answer depends on the name alone,
+// so it tells the caller nothing about which buckets exist. Authentication
+// failures (no identity resolved) and every other answer are left unchanged.
+func refineDenialForBucketName(r *http.Request, identity *Identity, errCode s3err.ErrorCode) s3err.ErrorCode {
+	if errCode != s3err.ErrAccessDenied || identity == nil {
+		return errCode
+	}
+	bucket, _ := s3_constants.GetBucketAndObject(r)
+	if bucket == "" {
+		return errCode
+	}
+	if !isASCIIBucketName(bucket) || s3bucket.VerifyS3BucketName(bucket) != nil {
+		return s3err.ErrInvalidBucketName
+	}
+	return errCode
+}
+
+// isASCIIBucketName reports whether name uses only the characters S3 allows
+// in a bucket name. VerifyS3BucketName checks with unicode.IsLower and
+// unicode.IsNumber, which also accept letters such as 'ü' and digits such as
+// '²' that S3 refuses.
+func isASCIIBucketName(name string) bool {
+	for i := 0; i < len(name); i++ {
+		c := name[i]
+		if !(c >= 'a' && c <= 'z' || c >= '0' && c <= '9' || c == '.' || c == '-') {
+			return false
+		}
+	}
+	return true
+}
+
 // recordIdentityInContext stores the authenticated identity, its name and its
 // principal ARN in the request context. An STS session's name is only an opaque
 // session subject, so the ARN is what carries the assumed role and session name

weed/s3api/auth_invalid_bucket_name_test.go

@@ -0,0 +1,250 @@
+package s3api
+
+import (
+	"bytes"
+	"encoding/base64"
+	"fmt"
+	"mime/multipart"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/gorilla/mux"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	invalidNameTestAccessKey = "INVALIDNAMEKEY000001"
+	invalidNameTestSecretKey = "invalid-name-secret-for-tests-only"
+	invalidNameTestBucket    = "readable-bucket"
+)
+
+// newInvalidNameIAM has one identity that may read invalidNameTestBucket only
+// and a configured anonymous identity with no actions.
+func newInvalidNameIAM(t *testing.T) *IdentityAccessManagement {
+	t.Helper()
+	iam := &IdentityAccessManagement{hashes: make(map[string]*sync.Pool), hashCounters: make(map[string]*int32)}
+	require.NoError(t, iam.loadS3ApiConfiguration(&iam_pb.S3ApiConfiguration{
+		Identities: []*iam_pb.Identity{
+			{Name: "reader", Actions: []string{"Read:" + invalidNameTestBucket},
+				Credentials: []*iam_pb.Credential{{AccessKey: invalidNameTestAccessKey, SecretKey: invalidNameTestSecretKey}}},
+			{Name: s3_constants.AccountAnonymousId},
+		},
+	}))
+	require.True(t, iam.isEnabled())
+	return iam
+}
+
+func invalidNameRequest(t *testing.T, bucket string, sign bool) *http.Request {
+	t.Helper()
+	req := mustNewRequest(http.MethodGet, fmt.Sprintf("http://127.0.0.1:8333/%s/key", bucket), 0, nil, t)
+	if sign {
+		require.NoError(t, signRequestV4(req, invalidNameTestAccessKey, invalidNameTestSecretKey))
+	}
+	return mux.SetURLVars(req, map[string]string{"bucket": bucket, "object": "key"})
+}
+
+func serveAuth(iam *IdentityAccessManagement, r *http.Request) (int, bool) {
+	ran := false
+	rec := httptest.NewRecorder()
+	iam.Auth(func(w http.ResponseWriter, r *http.Request) { ran = true; w.WriteHeader(http.StatusOK) }, s3_constants.ACTION_READ).ServeHTTP(rec, r)
+	return rec.Code, ran
+}
+
+// A denied request on a malformed bucket name answers 400 InvalidBucketName,
+// as AWS S3 does; a denied request on a well-formed name stays 403.
+func TestAuthDeniedInvalidBucketName(t *testing.T) {
+	iam := newInvalidNameIAM(t)
+
+	code, ran := serveAuth(iam, invalidNameRequest(t, invalidNameTestBucket, true))
+	assert.True(t, ran, "the identity's own bucket is allowed")
+	assert.Equal(t, http.StatusOK, code)
+
+	code, ran = serveAuth(iam, invalidNameRequest(t, "other-bucket", true))
+	assert.False(t, ran)
+	assert.Equal(t, http.StatusForbidden, code, "a well-formed name the caller may not read stays AccessDenied")
+
+	for _, bucket := range []string{"Bad_Bucket", "ab", "bucket..name", "bücket"} {
+		for _, sign := range []bool{true, false} {
+			code, ran = serveAuth(iam, invalidNameRequest(t, bucket, sign))
+			assert.False(t, ran, bucket)
+			assert.Equal(t, http.StatusBadRequest, code, "bucket %q signed=%v", bucket, sign)
+		}
+	}
+}
+
+func TestRefineDenialForBucketName(t *testing.T) {
+	reader := &Identity{Name: "reader"}
+	req := func(bucket string) *http.Request {
+		return mux.SetURLVars(httptest.NewRequest(http.MethodGet, "/", nil), map[string]string{"bucket": bucket})
+	}
+	cases := []struct {
+		name     string
+		bucket   string
+		identity *Identity
+		in, want s3err.ErrorCode
+	}{
+		{"space in the name", "bucket name", reader, s3err.ErrAccessDenied, s3err.ErrInvalidBucketName},
+		{"non-ASCII letter", "bücket", reader, s3err.ErrAccessDenied, s3err.ErrInvalidBucketName},
+		{"superscript digit", "b²cket", reader, s3err.ErrAccessDenied, s3err.ErrInvalidBucketName},
+		{"too short", "ab", reader, s3err.ErrAccessDenied, s3err.ErrInvalidBucketName},
+		{"well-formed name", "other-bucket", reader, s3err.ErrAccessDenied, s3err.ErrAccessDenied},
+		{"no bucket in the request", "", reader, s3err.ErrAccessDenied, s3err.ErrAccessDenied},
+		{"allowed", "Bad_Bucket", reader, s3err.ErrNone, s3err.ErrNone},
+		{"authentication failed", "Bad_Bucket", nil, s3err.ErrAccessDenied, s3err.ErrAccessDenied},
+		{"bad signature", "Bad_Bucket", nil, s3err.ErrSignatureDoesNotMatch, s3err.ErrSignatureDoesNotMatch},
+	}
+	for _, c := range cases {
+		assert.Equal(t, c.want, refineDenialForBucketName(req(c.bucket), c.identity, c.in), c.name)
+	}
+}
+
+// signedPostForm builds a multipart POST Object request for bucket, its policy
+// signed with secret under postAuthzAccessKey.
+func signedPostForm(t *testing.T, bucket, key, secret string) *http.Request {
+	t.Helper()
+	now := time.Now().UTC()
+	amzDate := now.Format(iso8601Format)
+	day := now.Format(yyyymmdd)
+	credential := fmt.Sprintf("%s/%s/%s/%s/aws4_request", postAuthzAccessKey, day, postAuthzRegion, postAuthzService)
+	policyJSON := fmt.Sprintf(`{"expiration":"%s","conditions":[["eq","$bucket","%s"],["eq","$key","%s"],`+
+		`["eq","$x-amz-credential","%s"],["eq","$x-amz-algorithm","AWS4-HMAC-SHA256"],["eq","$x-amz-date","%s"]]}`,
+		now.Add(time.Hour).Format("2006-01-02T15:04:05.000Z"), bucket, key, credential, amzDate)
+	encodedPolicy := base64.StdEncoding.EncodeToString([]byte(policyJSON))
+	signature := getSignature(getSigningKey(secret, day, postAuthzRegion, postAuthzService), encodedPolicy)
+
+	var buf bytes.Buffer
+	writer := multipart.NewWriter(&buf)
+	for _, field := range [][2]string{{"bucket", bucket}, {"key", key}, {"x-amz-credential", credential},
+		{"x-amz-algorithm", "AWS4-HMAC-SHA256"}, {"x-amz-date", amzDate}, {"policy", encodedPolicy}, {"x-amz-signature", signature}} {
+		require.NoError(t, writer.WriteField(field[0], field[1]))
+	}
+	filePart, err := writer.CreateFormFile("file", "payload.txt")
+	require.NoError(t, err)
+	_, err = filePart.Write([]byte("payload"))
+	require.NoError(t, err)
+	require.NoError(t, writer.Close())
+
+	req := httptest.NewRequest(http.MethodPost, "/"+bucket, &buf)
+	req.Header.Set("Content-Type", writer.FormDataContentType())
+	return mux.SetURLVars(req, map[string]string{"bucket": bucket})
+}
+
+// A browser POST upload authenticates in the handler, after AuthPostPolicy has
+// delegated it, so its denials are refined there: a verified form denied on a
+// malformed bucket name answers 400, a bad signature or a well-formed name 403.
+func TestPostPolicyDeniedInvalidBucketName(t *testing.T) {
+	s3a := newPostAuthzServer(t)
+	for _, bucket := range []string{"Bad_Bucket", "other-bucket"} {
+		s3a.bucketRegistry.setMetadataCache(&BucketMetaData{Name: bucket})
+	}
+	// A second identity may write Bad_Bucket, except under denied/.
+	require.NoError(t, s3a.iam.loadS3ApiConfiguration(&iam_pb.S3ApiConfiguration{
+		Identities: []*iam_pb.Identity{
+			{Name: "tester", Actions: []string{"Write:" + postAuthzBucket},
+				Credentials: []*iam_pb.Credential{{AccessKey: postAuthzAccessKey, SecretKey: postAuthzSecretKey}}},
+		},
+	}))
+	require.NoError(t, s3a.policyEngine.engine.SetBucketPolicy("Bad_Bucket", `{"Version":"2012-10-17","Statement":[`+
+		`{"Effect":"Deny","Principal":"*","Action":["s3:PutObject"],"Resource":["arn:aws:s3:::Bad_Bucket/denied/*"]}]}`))
+
+	serve := func(r *http.Request) *httptest.ResponseRecorder {
+		rec := httptest.NewRecorder()
+		s3a.iam.AuthPostPolicy(s3a.PostPolicyBucketHandler, s3_constants.ACTION_WRITE).ServeHTTP(rec, r)
+		return rec
+	}
+
+	rec := serve(signedPostForm(t, "Bad_Bucket", "k", postAuthzSecretKey))
+	assert.Equal(t, http.StatusBadRequest, rec.Code, rec.Body.String())
+	assert.Contains(t, rec.Body.String(), "InvalidBucketName")
+
+	rec = serve(signedPostForm(t, "Bad_Bucket", "k", "not-the-secret"))
+	assert.Equal(t, http.StatusForbidden, rec.Code, rec.Body.String())
+	assert.Contains(t, rec.Body.String(), "SignatureDoesNotMatch", "a form that fails verification is not refined")
+
+	rec = serve(signedPostForm(t, "other-bucket", "k", postAuthzSecretKey))
+	assert.Equal(t, http.StatusForbidden, rec.Code, rec.Body.String())
+	assert.Contains(t, rec.Body.String(), "AccessDenied")
+
+	// The per-key authorization after the coarse Write check is refined too.
+	require.NoError(t, s3a.iam.loadS3ApiConfiguration(&iam_pb.S3ApiConfiguration{
+		Identities: []*iam_pb.Identity{
+			{Name: "tester", Actions: []string{"Write:Bad_Bucket"},
+				Credentials: []*iam_pb.Credential{{AccessKey: postAuthzAccessKey, SecretKey: postAuthzSecretKey}}},
+		},
+	}))
+	rec = serve(signedPostForm(t, "Bad_Bucket", "denied/k", postAuthzSecretKey))
+	assert.Equal(t, http.StatusBadRequest, rec.Code, rec.Body.String())
+	assert.Contains(t, rec.Body.String(), "InvalidBucketName")
+}
+
+// Unsigned GET/HEAD object reads are authorized inside the object handlers, not
+// by Auth; their denial on a malformed bucket name answers 400 as well.
+func TestAnonymousReadDeniedInvalidBucketName(t *testing.T) {
+	read := func(s3a *S3ApiServer, method, bucket string) int {
+		r := httptest.NewRequest(method, "/"+bucket+"/key", nil)
+		r = mux.SetURLVars(r, map[string]string{"bucket": bucket, "object": "key"})
+		handler := s3a.GetObjectHandler
+		if method == http.MethodHead {
+			handler = s3a.HeadObjectHandler
+		}
+		rr := httptest.NewRecorder()
+		s3a.AuthWithPublicRead(handler, s3_constants.ACTION_READ)(rr, r)
+		return rr.Code
+	}
+	for _, method := range []string{http.MethodGet, http.MethodHead} {
+		s3a, _ := newAnonymousReadTestServer(t)
+		s3a.iam.identityAnonymous = &Identity{Name: s3_constants.AccountAnonymousId}
+		assert.Equal(t, http.StatusBadRequest, read(s3a, method, "Bad_Bucket"), method)
+		assert.Equal(t, http.StatusForbidden, read(s3a, method, "other-bucket"), "%s: a well-formed missing bucket stays masked", method)
+
+		// No anonymous identity configured: no identity resolves, as in Auth.
+		s3a, _ = newAnonymousReadTestServer(t)
+		assert.Equal(t, http.StatusForbidden, read(s3a, method, "Bad_Bucket"), method)
+	}
+}
+
+// An existing private key and a missing key in the same malformed-name bucket
+// must answer alike, or the status code tells an anonymous caller which keys
+// exist.
+func TestAnonymousReadDeniedInvalidBucketNameExistingKey(t *testing.T) {
+	const bucket = "bücket"
+	read := func(s3a *S3ApiServer, method, object string) int {
+		r := httptest.NewRequest(method, "/"+bucket+"/"+object, nil)
+		r = mux.SetURLVars(r, map[string]string{"bucket": bucket, "object": object})
+		handler := s3a.GetObjectHandler
+		if method == http.MethodHead {
+			handler = s3a.HeadObjectHandler
+		}
+		rr := httptest.NewRecorder()
+		s3a.AuthWithPublicRead(handler, s3_constants.ACTION_READ)(rr, r)
+		return rr.Code
+	}
+	setup := func(anonymous bool) *S3ApiServer {
+		s3a, f := newAnonymousReadTestServer(t)
+		if anonymous {
+			s3a.iam.identityAnonymous = &Identity{Name: s3_constants.AccountAnonymousId}
+		}
+		s3a.bucketConfigCache.Set(bucket, &BucketConfig{Name: bucket, Ownership: s3_constants.OwnershipObjectWriter})
+		f.entries["/buckets/"+bucket+"/private"] = anonymousReadEntry([]byte(`[]`))
+		f.entries["/buckets/"+bucket+"/public"] = anonymousReadEntry(anonymousReadGrant("READ"))
+		return s3a
+	}
+	for _, method := range []string{http.MethodGet, http.MethodHead} {
+		s3a := setup(true)
+		assert.Equal(t, http.StatusOK, read(s3a, method, "public"), "%s: a public key is still served", method)
+		assert.Equal(t, http.StatusBadRequest, read(s3a, method, "missing"), method)
+		assert.Equal(t, http.StatusBadRequest, read(s3a, method, "private"), "%s: an existing private key answers like a missing one", method)
+
+		// No anonymous identity configured: both stay 403, as in Auth.
+		s3a = setup(false)
+		assert.Equal(t, http.StatusForbidden, read(s3a, method, "missing"), method)
+		assert.Equal(t, http.StatusForbidden, read(s3a, method, "private"), method)
+	}
+}

weed/s3api/auth_signature_v2.go

@@ -29,7 +29,6 @@ import (
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
-	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
 )
 
@@ -95,11 +94,6 @@ func (iam *IdentityAccessManagement) doesPolicySignatureV2Match(formValues http.
 		return nil, s3err.ErrAccessDenied
 	}
 
-	bucket := formValues.Get("bucket")
-	if !identity.CanDo(s3_constants.ACTION_WRITE, bucket, "") {
-		return nil, s3err.ErrAccessDenied
-	}
-
 	policy := formValues.Get("Policy")
 	if policy == "" {
 		return nil, s3err.ErrMissingFields

weed/s3api/auth_signature_v4.go

@@ -858,11 +858,6 @@ func (iam *IdentityAccessManagement) doesPolicySignatureV4Match(formValues http.
 		return nil, s3err.ErrAccessDenied
 	}
 
-	bucket := formValues.Get("bucket")
-	if !identity.CanDo(s3_constants.ACTION_WRITE, bucket, "") {
-		return nil, s3err.ErrAccessDenied
-	}
-
 	// Get signing key.
 	signingKey := getSigningKey(cred.SecretKey, credHeader.scope.date.Format(yyyymmdd), credHeader.scope.region, credHeader.scope.service)
 

weed/s3api/bucket_paths.go

@@ -2,8 +2,12 @@ package s3api
 
 import (
 	"errors"
+	"net/http"
 	"path"
+	"regexp"
+	"strconv"
 	"strings"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -83,6 +87,117 @@ func (s3a *S3ApiServer) validateTableBucketObjectPath(bucket, object string) err
 	return nil
 }
 
+// volumeVerifyKey matches the object Snowflake writes, reads, lists and
+// deletes at the base of an S3-compatible external volume before it creates
+// or writes Iceberg tables there (SYSTEM$VERIFY_EXTERNAL_VOLUME, also run
+// implicitly): <base>/verify_<epoch milliseconds>_<number>, a 21-byte body.
+// A table bucket refuses every key outside the Iceberg layout, so without
+// this exception Snowflake cannot use a table bucket as an external volume.
+//
+// The exception is narrow so that it cannot be used to store data:
+//   - only at the bucket root: nothing can be encoded in a prefix, written
+//     into a hidden tree or take a namespace or table name;
+//   - one shape: 13 digits of epoch milliseconds, then 1-10 digits;
+//   - never a directory: a namespace may carry a name of that shape, and the
+//     key then names its directory (volumeVerifyKeyIsDirectory);
+//   - a PUT must be dated within volumeVerifyWindow of the server's clock,
+//     declare a Content-Length of at most volumeVerifyMaxBytes (no chunked or
+//     undeclared body), and carry no user metadata, tagging, stored content
+//     headers or extraction request (volumeVerifyPutAllowed). Reads and
+//     deletes take a key of any date, so an object a check left behind can
+//     still be removed.
+//
+// Only the calls the check makes (PUT, GET, HEAD, DELETE, DeleteObjects)
+// accept the key; multipart, copy, rename, ACL, tagging and POST policy keep
+// the plain layout rule.
+var volumeVerifyKey = regexp.MustCompile(`^verify_([0-9]{13})_[0-9]{1,10}$`)
+
+const (
+	volumeVerifyMaxBytes = 1024
+	volumeVerifyWindow   = time.Hour
+)
+
+// volumeVerifyNow is the clock a PUT's key is checked against.
+var volumeVerifyNow = time.Now
+
+// volumeVerifyKeyTime is the time a verification key is dated, and whether
+// the object has the key's shape.
+func volumeVerifyKeyTime(object string) (time.Time, bool) {
+	m := volumeVerifyKey.FindStringSubmatch(strings.TrimPrefix(object, "/"))
+	if m == nil {
+		return time.Time{}, false
+	}
+	ms, err := strconv.ParseInt(m[1], 10, 64)
+	if err != nil {
+		return time.Time{}, false
+	}
+	return time.UnixMilli(ms), true
+}
+
+func isVolumeVerifyKey(object string) bool {
+	_, ok := volumeVerifyKeyTime(object)
+	return ok
+}
+
+// volumeVerifyLookup reads the entry a verification key names.
+var volumeVerifyLookup = func(s3a *S3ApiServer, dir, name string) (*filer_pb.Entry, error) {
+	return s3a.getEntry(dir, name)
+}
+
+// volumeVerifyKeyIsDirectory reports whether the key names a directory (a
+// namespace), or whether that cannot be known.
+func (s3a *S3ApiServer) volumeVerifyKeyIsDirectory(bucket, object string) bool {
+	entry, err := volumeVerifyLookup(s3a, s3a.bucketDir(bucket), strings.TrimPrefix(object, "/"))
+	if err != nil {
+		return !errors.Is(err, filer_pb.ErrNotFound)
+	}
+	return entry != nil && entry.IsDirectory
+}
+
+// validateTableBucketObjectPathOrVerify is validateTableBucketObjectPath that
+// also accepts the external-volume verification key, and reports it so the
+// PUT handler can apply its limits.
+func (s3a *S3ApiServer) validateTableBucketObjectPathOrVerify(bucket, object string) (isVerify bool, err error) {
+	if !s3a.isTableBucket(bucket) {
+		return false, nil
+	}
+	if isVolumeVerifyKey(object) && !s3a.volumeVerifyKeyIsDirectory(bucket, object) {
+		return true, nil
+	}
+	return false, s3a.validateTableBucketObjectPath(bucket, object)
+}
+
+// volumeVerifyPutAllowed reports whether a PUT to the verification key is
+// dated now, declares a small body and stores nothing beside it.
+func volumeVerifyPutAllowed(r *http.Request, object string) bool {
+	at, ok := volumeVerifyKeyTime(object)
+	if !ok {
+		return false
+	}
+	if d := volumeVerifyNow().Sub(at); d > volumeVerifyWindow || d < -volumeVerifyWindow {
+		return false
+	}
+	if r.ContentLength < 0 || r.ContentLength > volumeVerifyMaxBytes || len(r.TransferEncoding) > 0 {
+		return false
+	}
+	if isExtractZip(r) {
+		return false
+	}
+	for name, values := range r.Header {
+		n := strings.ToLower(name)
+		if strings.HasPrefix(n, "x-amz-meta-") || n == "x-amz-tagging" || n == "content-disposition" ||
+			n == "response-content-disposition" || n == "content-language" ||
+			n == "cache-control" || n == "expires" || n == "x-amz-website-redirect-location" {
+			return false
+		}
+		// aws-chunked only frames the body; any other coding is stored.
+		if n == "content-encoding" && storedContentEncoding(values) != "" {
+			return false
+		}
+	}
+	return true
+}
+
 func (s3a *S3ApiServer) bucketPrefix(bucket string) string {
 	return s3a.bucketDir(bucket) + "/"
 }

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/iceberg/handlers_commit.go

@@ -397,7 +397,7 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) {
 		if cleanupErr := s.deleteMetadataFile(r.Context(), metadataBucket, metadataPath, metadataFileName); cleanupErr != nil {
 			glog.V(1).Infof("Iceberg: failed to cleanup metadata file %s after update failure: %v", newMetadataLocation, cleanupErr)
 		}
-		glog.Errorf("Iceberg: CommitTable UpdateTable error: %v", err)
+		logManagerError("CommitTable UpdateTable", err)
 		writeManagerError(w, err)
 		return
 	}

weed/s3api/iceberg/handlers_namespace.go

@@ -140,7 +140,7 @@ func (s *Server) handleCreateNamespace(w http.ResponseWriter, r *http.Request) {
 			writeError(w, http.StatusConflict, "AlreadyExistsException", err.Error())
 			return
 		}
-		glog.Errorf("Iceberg: CreateNamespace error: %v", err)
+		logManagerError("CreateNamespace", err)
 		writeManagerError(w, err)
 		return
 	}

weed/s3api/iceberg/handlers_transaction.go

@@ -109,7 +109,7 @@ func (s *Server) handleCommitTransaction(w http.ResponseWriter, r *http.Request)
 				writeError(w, http.StatusConflict, "CommitFailedException", "Version token mismatch")
 				return
 			}
-			glog.Errorf("Iceberg: CommitTransaction UpdateTable error: %v", err)
+			logManagerError("CommitTransaction UpdateTable", err)
 			writeManagerError(w, err)
 			return
 		}

weed/s3api/iceberg/log_manager_error_test.go

@@ -0,0 +1,36 @@
+package iceberg
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
+)
+
+func TestLogManagerErrorRefusalsAreNotErrors(t *testing.T) {
+	var refusals, faults []string
+	oldR, oldF := logManagerRefusal, logManagerFault
+	logManagerRefusal = func(f string, a ...interface{}) { refusals = append(refusals, fmt.Sprintf(f, a...)) }
+	logManagerFault = func(f string, a ...interface{}) { faults = append(faults, fmt.Sprintf(f, a...)) }
+	defer func() { logManagerRefusal, logManagerFault = oldR, oldF }()
+	for _, c := range []struct {
+		err   error
+		fault bool
+	}{
+		{&s3tables.S3TablesError{Type: s3tables.ErrCodeAccessDenied, Message: "not authorized to create namespace in this bucket"}, false},
+		{&s3tables.S3TablesError{Type: s3tables.ErrCodeNoSuchNamespace, Message: "no such namespace"}, false},
+		{&s3tables.S3TablesError{Type: s3tables.ErrCodeNamespaceAlreadyExists, Message: "exists"}, false},
+		{errors.New("filer: connection refused"), true}, // positive control: a real fault stays at ERROR
+	} {
+		refusals, faults = nil, nil
+		logManagerError("CreateNamespace", c.err)
+		wantFaults, wantRefusals := 0, 1
+		if c.fault {
+			wantFaults, wantRefusals = 1, 0
+		}
+		if len(faults) != wantFaults || len(refusals) != wantRefusals {
+			t.Errorf("%v: %d ERROR and %d refusal log(s), want %d and %d (refusals %v, faults %v)", c.err, len(faults), len(refusals), wantFaults, wantRefusals, refusals, faults)
+		}
+	}
+}

weed/s3api/iceberg/utils.go

@@ -158,7 +158,9 @@ func nameValidationError(err error) bool {
 // errors (400) and must carry the reason -- a variant field without
 // format-version 3 otherwise reads as a bare 500 with "variant is not supported
 // until v3" only in the log. Anything else is a server fault (500).
-func writeManagerError(w http.ResponseWriter, err error) {
+// classifyManagerError maps a storage-layer error to the HTTP status, the
+// Iceberg error type and the message the catalog answers with.
+func classifyManagerError(err error) (int, string, string) {
 	switch {
 	case nameValidationError(err),
 		errors.Is(err, iceberg.ErrInvalidSchema),
@@ -166,8 +168,7 @@ func writeManagerError(w http.ResponseWriter, err error) {
 		errors.Is(err, iceberg.ErrInvalidTypeString),
 		errors.Is(err, iceberg.ErrInvalidTransform),
 		errors.Is(err, iceberg.ErrInvalidArgument):
-		writeError(w, http.StatusBadRequest, "BadRequestException", err.Error())
-		return
+		return http.StatusBadRequest, "BadRequestException", err.Error()
 	}
 	// Storage-layer failures are mostly the client's, not the server's. Reporting
 	// a missing namespace or a name conflict as a 500 makes the catalog look
@@ -179,31 +180,49 @@ func writeManagerError(w http.ResponseWriter, err error) {
 			// The storage-layer message names the resolved bucket, which for a
 			// client that sent no warehouse at all is the default one it never
 			// asked for, so say how to select a real table bucket.
-			writeError(w, http.StatusNotFound, "NoSuchNamespaceException",
-				fmt.Sprintf("%s: each table bucket is a separate catalog, select one with warehouse=s3://<table-bucket>/ or /v1/<table-bucket>/", tableErr.Message))
+			return http.StatusNotFound, "NoSuchNamespaceException",
+				fmt.Sprintf("%s: each table bucket is a separate catalog, select one with warehouse=s3://<table-bucket>/ or /v1/<table-bucket>/", tableErr.Message)
 		case s3tables.ErrCodeNoSuchNamespace:
-			writeError(w, http.StatusNotFound, "NoSuchNamespaceException", tableErr.Message)
+			return http.StatusNotFound, "NoSuchNamespaceException", tableErr.Message
 		case s3tables.ErrCodeNoSuchTable:
-			writeError(w, http.StatusNotFound, "NoSuchTableException", tableErr.Message)
+			return http.StatusNotFound, "NoSuchTableException", tableErr.Message
 		case s3tables.ErrCodeNoSuchView:
-			writeError(w, http.StatusNotFound, "NoSuchViewException", tableErr.Message)
+			return http.StatusNotFound, "NoSuchViewException", tableErr.Message
 		case s3tables.ErrCodeNamespaceAlreadyExists, s3tables.ErrCodeBucketAlreadyExists,
 			s3tables.ErrCodeTableAlreadyExists, s3tables.ErrCodeViewAlreadyExists:
-			writeError(w, http.StatusConflict, "AlreadyExistsException", tableErr.Message)
+			return http.StatusConflict, "AlreadyExistsException", tableErr.Message
 		case s3tables.ErrCodeNamespaceNotEmpty, s3tables.ErrCodeBucketNotEmpty:
-			writeError(w, http.StatusConflict, "AlreadyExistsException", tableErr.Message)
+			return http.StatusConflict, "AlreadyExistsException", tableErr.Message
 		case s3tables.ErrCodeConflict:
-			writeError(w, http.StatusConflict, "CommitFailedException", tableErr.Message)
+			return http.StatusConflict, "CommitFailedException", tableErr.Message
 		case s3tables.ErrCodeAccessDenied:
-			writeError(w, http.StatusForbidden, "ForbiddenException", tableErr.Message)
+			return http.StatusForbidden, "ForbiddenException", tableErr.Message
 		case s3tables.ErrCodeInvalidRequest, s3tables.ErrCodeInvalidIcebergLayout:
-			writeError(w, http.StatusBadRequest, "BadRequestException", tableErr.Message)
-		default:
-			writeError(w, http.StatusInternalServerError, "InternalServerError", err.Error())
+			return http.StatusBadRequest, "BadRequestException", tableErr.Message
 		}
+	}
+	return http.StatusInternalServerError, "InternalServerError", err.Error()
+}
+
+func writeManagerError(w http.ResponseWriter, err error) {
+	status, typ, msg := classifyManagerError(err)
+	writeError(w, status, typ, msg)
+}
+
+// A storage-layer error the catalog answers with a 4xx (access denied, not
+// found, already exists, not empty, a conflict) is the API refusing a request
+// as intended, not a server fault; it is logged at V(1). A 5xx stays at ERROR.
+var (
+	logManagerRefusal = func(format string, args ...interface{}) { glog.V(1).Infof(format, args...) }
+	logManagerFault   = glog.Errorf
+)
+
+func logManagerError(op string, err error) {
+	if status, _, _ := classifyManagerError(err); status >= 400 && status < 500 {
+		logManagerRefusal("Iceberg: %s refused (%d): %v", op, status, err)
 		return
 	}
-	writeError(w, http.StatusInternalServerError, "InternalServerError", err.Error())
+	logManagerFault("Iceberg: %s error: %v", op, err)
 }
 
 // resolveWarehouseBucket maps a client-supplied warehouse value to a table

weed/s3api/s3api_object_anonymous_read.go

@@ -171,10 +171,12 @@ func (s3a *S3ApiServer) authorizeAnonymousBucketList(r *http.Request, bucket str
 
 // writeAnonymousAwareNotFound keeps missing-bucket and missing-key misses
 // indistinguishable from denied reads for anonymous callers: a 404 only
-// reaches the wire when the bucket is anonymously listable.
+// reaches the wire when the bucket is anonymously listable. The denial is
+// refined like Auth's, so a malformed bucket name answers InvalidBucketName.
 func (s3a *S3ApiServer) writeAnonymousAwareNotFound(w http.ResponseWriter, r *http.Request, bucket string, code s3err.ErrorCode) {
 	if isAnonymousObjectRead(r) && s3a.authorizeAnonymousBucketList(r, bucket) != s3err.ErrNone {
-		s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
+		identity, _ := s3_constants.GetIdentityFromContext(r).(*Identity)
+		s3err.WriteErrorResponse(w, r, refineDenialForBucketName(r, identity, s3err.ErrAccessDenied))
 		return
 	}
 	s3err.WriteErrorResponse(w, r, code)

weed/s3api/s3api_object_anonymous_read_test.go

@@ -299,7 +299,9 @@ func TestAnonymousObjectACLConfiguredIdentity(t *testing.T) {
 	}{
 		{"legacy anonymous permission", false, []Action{s3_constants.ACTION_READ}, "", false, 200},
 		{"identity policy allow", false, nil, `{"Effect":"Allow","Action":"s3:GetObject","Resource":"arn:aws:s3:::b/*"}`, false, 200},
-		{"identity deny beats object ACL", true, nil, `{"Effect":"Deny","Action":"s3:GetObject","Resource":"arn:aws:s3:::b/*"}`, false, 403},
+		// The fixture bucket "b" is shorter than S3's three-character minimum,
+		// so a denial for a resolved identity answers InvalidBucketName.
+		{"identity deny beats object ACL", true, nil, `{"Effect":"Deny","Action":"s3:GetObject","Resource":"arn:aws:s3:::b/*"}`, false, 400},
 		{"disabled identity cannot grant", false, []Action{s3_constants.ACTION_READ}, "", true, 403},
 		{"public ACL does not require identity", true, nil, "", true, 200},
 	} {

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)
@@ -784,7 +804,7 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request)
 
 	bucket, object := s3_constants.GetBucketAndObject(r)
 	glog.V(3).Infof("GetObjectHandler %s %s", bucket, object)
-	if err := s3a.validateTableBucketObjectPath(bucket, object); err != nil {
+	if _, err := s3a.validateTableBucketObjectPathOrVerify(bucket, object); err != nil {
 		s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
 		return
 	}
@@ -2479,7 +2499,7 @@ func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request
 
 	bucket, object := s3_constants.GetBucketAndObject(r)
 	glog.V(3).Infof("HeadObjectHandler %s %s", bucket, object)
-	if err := s3a.validateTableBucketObjectPath(bucket, object); err != nil {
+	if _, err := s3a.validateTableBucketObjectPathOrVerify(bucket, object); err != nil {
 		s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
 		return
 	}
@@ -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,14 +1,21 @@
 package s3api
 
 import (
+	"bytes"
+	"crypto/md5"
+	"encoding/base64"
+	"fmt"
+	"net/http"
 	"net/http/httptest"
 	"strings"
 	"testing"
 	"time"
 
+	"github.com/gorilla/mux"
 	"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 +152,349 @@ 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")
+	}
+}
+
+// The handler must route an inline-Content source away from the raw chunk
+// copy — CopyObjectPartHandler has to send it through the plaintext stream
+// path or the part lands as zeros.
+func TestCopyObjectPartHandlerInlineContent(t *testing.T) {
+	volume := startFakeVolumeServer(t)
+	filerImpl := &ambiguousPutFiler{volume: volume, apply: true, entries: map[string]*filer_pb.Entry{}}
+	s3a := newPutTestServer(t, startFakeFiler(t, filerImpl))
+	s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
+	util_http.InitGlobalHttpClient()
+
+	content := []byte("inline-content")
+	uploadID := s3a.generateUploadID("dst")
+	filerImpl.entries["/buckets/b"] = &filer_pb.Entry{Name: "b", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}}
+	filerImpl.entries["/buckets/b/src"] = &filer_pb.Entry{
+		Name:       "src",
+		Attributes: &filer_pb.FuseAttributes{FileSize: uint64(len(content))},
+		Content:    content,
+	}
+	filerImpl.entries["/buckets/b/.uploads/"+uploadID] = &filer_pb.Entry{
+		Name:        uploadID,
+		IsDirectory: true,
+		Extended:    map[string][]byte{s3_constants.ExtMultipartObjectKey: []byte("dst")},
+	}
+
+	r := httptest.NewRequest(http.MethodPut, "/b/dst?partNumber=1&uploadId="+uploadID, nil)
+	r = mux.SetURLVars(r, map[string]string{"bucket": "b", "object": "dst"})
+	r.Header.Set("X-Amz-Copy-Source", "/b/src")
+	r.Header.Set("x-amz-copy-source-range", "bytes=2-5")
+	w := httptest.NewRecorder()
+	s3a.CopyObjectPartHandler(w, r)
+	if w.Code != http.StatusOK {
+		t.Fatalf("CopyObjectPartHandler = %d %s, want 200", w.Code, w.Body.String())
+	}
+	wantMd5 := fmt.Sprintf("%x", md5.Sum(content[2:6]))
+	if !strings.Contains(w.Body.String(), wantMd5) {
+		t.Errorf("response %q does not contain copied-range ETag %q", w.Body.String(), wantMd5)
+	}
+	var partEntry *filer_pb.Entry
+	for key, e := range filerImpl.entries {
+		if strings.Contains(key, uploadID) && !e.IsDirectory {
+			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_delete.go

@@ -210,7 +210,7 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque
 	// The filer ops below each retry, and a failover walk runs the whole set
 	// once per filer, so the backoff comes out of one allowance held here.
 	r = r.WithContext(withFilerRetryBudget(r.Context(), filerRetryRequestBudget))
-	if err := s3a.validateTableBucketObjectPath(bucket, object); err != nil {
+	if _, err := s3a.validateTableBucketObjectPathOrVerify(bucket, object); err != nil {
 		s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
 		return
 	}
@@ -479,7 +479,7 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
 				deleteErrors = append(deleteErrors, deleteErrorFromCode(validationCode, object.Key, object.VersionId))
 				continue
 			}
-			if err := s3a.validateTableBucketObjectPath(bucket, object.Key); err != nil {
+			if _, err := s3a.validateTableBucketObjectPathOrVerify(bucket, object.Key); err != nil {
 				deleteErrors = append(deleteErrors, deleteErrorFromCode(s3err.ErrAccessDenied, object.Key, object.VersionId))
 				continue
 			}

weed/s3api/s3api_object_handlers_list.go

@@ -540,18 +540,6 @@ func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, req listObjectsReq
 				}
 				dirName, entryName, _ := entryUrlEncode(dir, entry.Name, encodingTypeUrl)
 				if entry.IsDirectory {
-					if cursor.emitDirAsCommonPrefix {
-						cursor.emitDirAsCommonPrefix = false
-						dirPrefix := fmt.Sprintf("%s/%s/", dir, entry.Name)[len(bucketPrefix):]
-						commonPrefixes = append(commonPrefixes, PrefixEntry{
-							Prefix: dirPrefix,
-						})
-						cursor.maxKeys--
-						lastEntryWasCommonPrefix = true
-						lastCommonPrefix = dirPrefix
-						lastPrefixNullBackers, lastPrefixConfirmed = nil, true
-						return
-					}
 					if originalPrefix != "" {
 						normalizedPrefix := strings.TrimPrefix(strings.TrimSuffix(originalPrefix, "/"), "/")
 						if normalizedPrefix != "" {
@@ -806,9 +794,6 @@ type ListingCursor struct {
 	// resolvePendingNulls settles trailing null objects whose .versions sibling
 	// has not streamed yet before a page is declared full.
 	resolvePendingNulls func() error
-	// emitDirAsCommonPrefix marks the next directory entry passed to the entry
-	// callback for reporting as a CommonPrefix rather than a Contents key.
-	emitDirAsCommonPrefix bool
 	// prefixObjectKeyOnly limits the next prefix-object entry passed to the entry
 	// callback to its plain key; its "<name>/" CommonPrefix is deferred to the
 	// directory's own block position.
@@ -1024,10 +1009,8 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.
 		}
 		isKeyObject := e.IsDirectoryKeyObject() && !e.IsPrefixObject()
 		if delimiter != "/" || cursor.prefixEndsOnDelimiter {
-			// A trailing-slash prefix (e.g. "logs/") names one directory and asks
-			// whether it exists, so a real but empty directory must be reported for
-			// that probe.
-			explicitDirProbe := cursor.prefixEndsOnDelimiter
+			// A trailing-slash prefix (e.g. "logs/") names one directory, so the
+			// recursion below answers whether it exists and what it holds.
 			if cursor.prefixEndsOnDelimiter {
 				cursor.prefixEndsOnDelimiter = false
 			}
@@ -1036,13 +1019,7 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.
 				// appear as regular keys in recursive listing mode.
 				eachEntryFn(dir, emitEntry)
 			}
-			// Recurse into subdirectory to list any children, noting whether the
-			// subtree produced any entries.
-			childEmitted := false
-			subNextMarker, subErr := s3a.doListFilerEntries(ctx, client, listDirectoryRequest{dir: dir + "/" + e.Name, marker: subMarker, delimiter: delimiter, bucket: bucket}, cursor, func(d string, ent *filer_pb.Entry) {
-				childEmitted = true
-				eachEntryFn(d, ent)
-			})
+			subNextMarker, subErr := s3a.doListFilerEntries(ctx, client, listDirectoryRequest{dir: dir + "/" + e.Name, marker: subMarker, delimiter: delimiter, bucket: bucket}, cursor, eachEntryFn)
 			if subErr != nil {
 				return fmt.Errorf("doListFilerEntries2: %w", subErr)
 			}
@@ -1052,26 +1029,10 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.
 				// keep a resume position that re-enters it.
 				cursor.isTruncated = true
 			}
-			// A real but empty directory (created out of band via mount, mkdir or
-			// the filer API, so it carries no MIME) is otherwise invisible to S3
-			// clients that detect directories by listing it under its own "<dir>/"
-			// prefix. Answer that explicit probe so tools like hadoop-aws can find
-			// the directory: a delimited listing reports it as a CommonPrefix --
-			// a real key in Contents would look like an object a client can read --
-			// while a recursive listing keeps the directory marker key it used.
-			// Plain listings are left untouched, so empty directories left behind
-			// by deleted objects are not shown as keys.
-			// A directory that still holds version history no longer names anything,
-			// so it gets no marker either.
-			if explicitDirProbe && !isKeyObject && !childEmitted && !cursor.isTruncated && e.Attributes != nil &&
-				!s3a.dirHoldsOnlyHiddenEntries(ctx, client, bucket, dir+"/"+e.Name, cursor) {
-				if delimiter == "/" {
-					cursor.emitDirAsCommonPrefix = true
-				} else {
-					emitEntry.Attributes.Mime = s3_constants.FolderMimeType
-				}
-				eachEntryFn(dir, emitEntry)
-			}
+			// An empty directory gets no marker under its own "<dir>/" prefix:
+			// that listing enumerates children, where S3 returns nothing for an
+			// empty prefix, and a fake entry is read back as a real object by
+			// clients that enumerate locks or similar.
 			// println("doListFilerEntries2 dir", dir+"/"+e.Name, "subNextMarker", subNextMarker)
 			if cursor.isTruncated || matchedPrefixDir {
 				stopped = true

weed/s3api/s3api_object_handlers_list_directory_test.go

@@ -65,12 +65,11 @@ func TestDirectoryListedAsCommonPrefix(t *testing.T) {
 	assert.Empty(t, seenFiles, "No files should be seen, only the directory")
 }
 
-// TestEmptyDirectorySurfacedAsMarker reproduces the hadoop-aws / Spark case where a
-// real but empty directory (created via mount, mkdir or the filer API, so it has no
-// MIME) must be visible to S3 clients that detect directories by listing under the
-// "<dir>/" prefix. Under a delimiter it is reported as a CommonPrefix, which the
-// probe accepts, instead of a key a client might try to read.
-func TestEmptyDirectorySurfacedAsMarker(t *testing.T) {
+// TestEmptyDirectoryNotListedUnderOwnPrefix covers the explicit "<dir>/" probe
+// on a real but empty directory (created via mount, mkdir or the filer API, so
+// it has no MIME): it is listed under its own prefix in neither delimited nor
+// recursive mode, matching S3 where an empty prefix has no key to return.
+func TestEmptyDirectoryNotListedUnderOwnPrefix(t *testing.T) {
 	s3a := &S3ApiServer{
 		option: &S3ApiServerOption{BucketsPath: "/buckets"},
 	}
@@ -87,20 +86,15 @@ func TestEmptyDirectorySurfacedAsMarker(t *testing.T) {
 		},
 	}
 
-	// Mirrors the getFileStatus probe: prefix "logs/" with delimiter "/".
-	cursor := &ListingCursor{maxKeys: 1000, prefixEndsOnDelimiter: true}
-	var seen []*filer_pb.Entry
-	_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test", prefix: "logs", delimiter: "/", bucket: "test"}, cursor,
-		func(dir string, entry *filer_pb.Entry) {
-			seen = append(seen, entry)
-		})
-
-	assert.NoError(t, err)
-	if assert.Len(t, seen, 1, "empty directory must be surfaced under the logs/ prefix") {
-		assert.True(t, cursor.emitDirAsCommonPrefix,
-			"a delimited probe reports the empty directory as a common prefix")
-		assert.Empty(t, seen[0].Attributes.Mime,
-			"the entry must not be dressed as a directory key object")
+	for _, delimiter := range []string{"/", ""} {
+		cursor := &ListingCursor{maxKeys: 1000, prefixEndsOnDelimiter: true}
+		var seen []*filer_pb.Entry
+		_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test", prefix: "logs", delimiter: delimiter, bucket: "test"}, cursor,
+			func(dir string, entry *filer_pb.Entry) {
+				seen = append(seen, entry)
+			})
+		assert.NoError(t, err)
+		assert.Empty(t, seen, "an empty directory reports nothing under its own prefix (delimiter %q)", delimiter)
 	}
 }
 
@@ -185,10 +179,10 @@ func TestEmptyDirectoryHiddenInFlatListing(t *testing.T) {
 	assert.Equal(t, []string{"foo"}, seen, "flat listing must hide the empty directory and return only real objects")
 }
 
-// A trailing-slash delimited listing of an empty directory reports the
-// directory as a CommonPrefix, not as a Contents key: clients that treat
-// every returned key as an object (e.g. lock listings) skip prefixes.
-func TestEmptyDirectoryListedAsCommonPrefix(t *testing.T) {
+// A trailing-slash delimited listing of an empty directory enumerates its
+// children, so it returns nothing: the directory is neither a readable Contents
+// key nor a CommonPrefix of its own.
+func TestEmptyDirectoryHiddenInDelimitedSelfListing(t *testing.T) {
 	filer := &fakeBucketDeleteFiler{entriesByDir: map[string][]*filer_pb.Entry{
 		"/buckets/test": {
 			{Name: "locks", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mime: ""}},
@@ -208,6 +202,5 @@ func TestEmptyDirectoryListedAsCommonPrefix(t *testing.T) {
 	})
 	require.NoError(t, err)
 	assert.Empty(t, response.Contents, "an empty directory is not a readable key")
-	require.Len(t, response.CommonPrefixes, 1)
-	assert.Equal(t, "locks/", response.CommonPrefixes[0].Prefix)
+	assert.Empty(t, response.CommonPrefixes, "an empty directory has no prefixes to fold into")
 }

weed/s3api/s3api_object_handlers_postpolicy.go

@@ -82,7 +82,7 @@ func (s3a *S3ApiServer) PostPolicyBucketHandler(w http.ResponseWriter, r *http.R
 	// Verify policy signature.
 	identity, errCode := s3a.iam.doesPolicySignatureMatch(formValues)
 	if errCode != s3err.ErrNone {
-		s3err.WriteErrorResponse(w, r, errCode)
+		s3err.WriteErrorResponse(w, r, refineDenialForBucketName(r, identity, errCode))
 		return
 	}
 	if identity != nil {
@@ -140,7 +140,7 @@ func (s3a *S3ApiServer) PostPolicyBucketHandler(w http.ResponseWriter, r *http.R
 
 	// Authorize the object like the PUT path; the coarse Write check above does not consult the bucket policy.
 	if errCode := s3a.iam.AuthorizeObjectWrite(r, identity, bucket, object); errCode != s3err.ErrNone {
-		s3err.WriteErrorResponse(w, r, errCode)
+		s3err.WriteErrorResponse(w, r, refineDenialForBucketName(r, identity, errCode))
 		return
 	}
 
@@ -362,10 +362,24 @@ func getRedirectPostRawQuery(bucket, key, etag string) string {
 }
 
 // Check to see if Policy is signed correctly, returning the signing identity.
+// The identity's Write permission on the bucket is checked only once the
+// signature verifies, as on the header-signed path, so a form that fails
+// verification learns nothing about the key's permissions. A verified identity
+// that may not write is returned with ErrAccessDenied.
 func (iam *IdentityAccessManagement) doesPolicySignatureMatch(formValues http.Header) (*Identity, s3err.ErrorCode) {
+	var identity *Identity
+	var errCode s3err.ErrorCode
 	// For SignV2 - Signature field will be valid
 	if _, ok := formValues["Signature"]; ok {
-		return iam.doesPolicySignatureV2Match(formValues)
+		identity, errCode = iam.doesPolicySignatureV2Match(formValues)
+	} else {
+		identity, errCode = iam.doesPolicySignatureV4Match(formValues)
 	}
-	return iam.doesPolicySignatureV4Match(formValues)
+	if errCode != s3err.ErrNone {
+		return nil, errCode
+	}
+	if !identity.CanDo(s3_constants.ACTION_WRITE, formValues.Get("bucket"), "") {
+		return identity, s3err.ErrAccessDenied
+	}
+	return identity, s3err.ErrNone
 }

weed/s3api/s3api_object_handlers_put.go

@@ -112,7 +112,7 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 		s3err.WriteErrorResponse(w, r, s3err.ErrKeyTooLongError)
 		return
 	}
-	if err := s3a.validateTableBucketObjectPath(bucket, object); err != nil {
+	if isVerify, err := s3a.validateTableBucketObjectPathOrVerify(bucket, object); err != nil || (isVerify && !volumeVerifyPutAllowed(r, object)) {
 		s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
 		return
 	}
@@ -991,7 +991,11 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 			}
 			glog.V(3).Infof("putToFiler: Calling CreateEntry for %s", filePath)
 			err := filer_pb.CreateEntry(context.Background(), client, req)
-			if errors.Is(err, filer_pb.ErrExistingIsDirectory) && !isReservedDirectoryName(entry.Name) {
+			// A verification key in a table bucket never becomes a prefix
+			// object: the directory it meets is a namespace, created after the
+			// handler looked.
+			if errors.Is(err, filer_pb.ErrExistingIsDirectory) && !isReservedDirectoryName(entry.Name) &&
+				!(isVolumeVerifyKey(object) && s3a.isTableBucket(bucket)) {
 				// Other keys are nested under this one. S3 keys are flat, so the key is
 				// stored on the directory they live under rather than refused.
 				entry.MarkPrefixObject()

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/s3api_server.go

@@ -661,7 +661,10 @@ func (s3a *S3ApiServer) checkPolicyWithEntry(r *http.Request, bucket, object, ac
 // Returns s3err.ErrNone if allowed, or an error code if denied or on error.
 func (s3a *S3ApiServer) recheckPolicyWithObjectEntry(r *http.Request, bucket, object, action string, objectEntry map[string][]byte, handlerName string) s3err.ErrorCode {
 	if isAnonymousObjectRead(r) {
-		return s3a.authorizeAnonymousObjectRead(r, bucket, object, objectEntry)
+		// Refined like writeAnonymousAwareNotFound's masked miss, so a denied
+		// existing key and a missing key answer alike on a malformed name.
+		identity, _ := GetIdentityFromContext(r).(*Identity)
+		return refineDenialForBucketName(r, identity, s3a.authorizeAnonymousObjectRead(r, bucket, object, objectEntry))
 	}
 	identityRaw := GetIdentityFromContext(r)
 	var identity *Identity

weed/s3api/s3tables/handler.go

@@ -127,6 +127,10 @@ func (h *S3TablesHandler) HandleRequest(w http.ResponseWriter, r *http.Request,
 
 	glog.V(3).Infof("S3Tables: handling operation %s", operation)
 
+	// the status the handler answers decides how its error is logged (logRequestError)
+	sw := &statusRecorder{ResponseWriter: w}
+	w = sw
+
 	var err error
 	switch operation {
 	// Table Bucket operations
@@ -223,10 +227,51 @@ func (h *S3TablesHandler) HandleRequest(w http.ResponseWriter, r *http.Request,
 	}
 
 	if err != nil {
-		glog.Errorf("S3Tables: error handling %s: %v", operation, err)
+		logRequestError(operation, sw.status, err)
 	}
 }
 
+// A refusal the handler answered with a 4xx (access denied, not authorized,
+// not found, already exists, not empty) is the API working as intended, not a
+// server fault; it is logged at V(1). A 5xx, or an error returned without any
+// status written, stays at ERROR.
+var (
+	logClientRefusal = func(format string, args ...interface{}) { glog.V(1).Infof(format, args...) }
+	logServerError   = glog.Errorf
+)
+
+func logRequestError(operation string, status int, err error) {
+	if status >= 400 && status < 500 {
+		logClientRefusal("S3Tables: %s refused (%d): %v", operation, status, err)
+		return
+	}
+	logServerError("S3Tables: error handling %s: %v", operation, err)
+}
+
+// statusRecorder remembers the first status written through it.
+type statusRecorder struct {
+	http.ResponseWriter
+	status int
+}
+
+func (s *statusRecorder) WriteHeader(code int) {
+	if s.status == 0 {
+		s.status = code
+	}
+	s.ResponseWriter.WriteHeader(code)
+}
+
+func (s *statusRecorder) Write(b []byte) (int, error) {
+	if s.status == 0 {
+		s.status = http.StatusOK
+	}
+	return s.ResponseWriter.Write(b)
+}
+
+// Unwrap lets http.NewResponseController reach the underlying writer, so it
+// flushes (or reports ErrNotSupported) exactly as that writer does.
+func (s *statusRecorder) Unwrap() http.ResponseWriter { return s.ResponseWriter }
+
 // Principal/authorization helpers
 
 // getAccountID returns a stable caller identifier for ownership and permission checks.
@@ -459,6 +504,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_log_level_test.go

@@ -0,0 +1,123 @@
+package s3tables
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+)
+
+// captureLogs swaps the two log sinks for recorders and restores them when the test ends.
+func captureLogs(t *testing.T) (refusals, serverErrors *[]string) {
+	t.Helper()
+	var r, s []string
+	oldR, oldS := logClientRefusal, logServerError
+	logClientRefusal = func(f string, a ...interface{}) { r = append(r, fmt.Sprintf(f, a...)) }
+	logServerError = func(f string, a ...interface{}) { s = append(s, fmt.Sprintf(f, a...)) }
+	t.Cleanup(func() { logClientRefusal, logServerError = oldR, oldS })
+	return &r, &s
+}
+
+// A refusal the API answers with a 4xx is not a server error.
+func TestRequestRefusedWith4xxIsNotLoggedAsError(t *testing.T) {
+	refusals, serverErrors := captureLogs(t)
+	h := NewS3TablesHandler()
+	req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader("not json"))
+	req.Header.Set("X-Amz-Target", "S3Tables.ListNamespaces")
+	rec := httptest.NewRecorder()
+	h.HandleRequest(rec, req, nil)
+	if rec.Code < 400 || rec.Code >= 500 {
+		t.Fatalf("the probe request must be refused with a 4xx for this test to mean anything, got %d", rec.Code) // positive control
+	}
+	if len(*serverErrors) != 0 {
+		t.Fatalf("a %d refusal was logged at ERROR: %v", rec.Code, *serverErrors)
+	}
+	if len(*refusals) != 1 || !strings.Contains((*refusals)[0], "ListNamespaces refused") {
+		t.Fatalf("the refusal must still be logged once (at V(1)), got %v", *refusals)
+	}
+}
+
+func TestLogRequestErrorByStatus(t *testing.T) {
+	for _, c := range []struct {
+		status    int
+		wantError bool
+	}{{http.StatusForbidden, false}, {http.StatusNotFound, false}, {http.StatusConflict, false},
+		{http.StatusInternalServerError, true}, {http.StatusServiceUnavailable, true}, {0, true}} {
+		refusals, serverErrors := captureLogs(t)
+		logRequestError("Op", c.status, errors.New("boom"))
+		if gotError := len(*serverErrors) == 1 && len(*refusals) == 0; gotError != c.wantError {
+			t.Errorf("status %d: ERROR=%v, want %v (refusals %v, errors %v)", c.status, gotError, c.wantError, *refusals, *serverErrors)
+		}
+	}
+}
+
+func TestStatusRecorderKeepsTheFirstStatus(t *testing.T) {
+	rec := httptest.NewRecorder()
+	sw := &statusRecorder{ResponseWriter: rec}
+	_, _ = sw.Write([]byte("x"))
+	sw.WriteHeader(http.StatusInternalServerError)
+	if sw.status != http.StatusOK {
+		t.Fatalf("an implicit 200 by Write must win, got %d", sw.status)
+	}
+	sw2 := &statusRecorder{ResponseWriter: httptest.NewRecorder()}
+	sw2.WriteHeader(http.StatusForbidden)
+	if sw2.status != http.StatusForbidden {
+		t.Fatalf("got %d", sw2.status)
+	}
+}
+
+func TestStatusRecorderUnwrapsForResponseController(t *testing.T) {
+	sw := &statusRecorder{ResponseWriter: httptest.NewRecorder()}
+	if err := http.NewResponseController(sw).Flush(); err != nil {
+		t.Fatalf("ResponseController through the recorder: %v", err)
+	}
+	if sw.Unwrap() == nil {
+		t.Fatal("Unwrap must return the wrapped writer")
+	}
+}
+
+// A malformed CreateNamespace body is answered with a 400; neither the
+// dispatcher nor the handler itself may log it at ERROR.
+func TestMalformedCreateNamespaceIsNotLoggedAsError(t *testing.T) {
+	refusals, serverErrors := captureLogs(t)
+
+	before := glog.Stats.Error.Lines()
+	glog.Errorf("S3Tables test: positive control for the ERROR line counter")
+	if glog.Stats.Error.Lines() != before+1 {
+		t.Fatal("the ERROR line counter did not see a direct glog.Errorf") // positive control
+	}
+
+	h := NewS3TablesHandler()
+	req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader("not json"))
+	req.Header.Set("X-Amz-Target", "S3Tables.CreateNamespace")
+	rec := httptest.NewRecorder()
+	before = glog.Stats.Error.Lines()
+	h.HandleRequest(rec, req, nil)
+	errorLines := glog.Stats.Error.Lines() - before
+	if rec.Code != http.StatusBadRequest {
+		t.Fatalf("a malformed body must be refused with 400, got %d", rec.Code)
+	}
+	if errorLines != 0 || len(*serverErrors) != 0 {
+		t.Fatalf("a 400 refusal was logged at ERROR: %d direct line(s), dispatcher %v", errorLines, *serverErrors)
+	}
+	if len(*refusals) != 1 || !strings.Contains((*refusals)[0], "CreateNamespace refused") {
+		t.Fatalf("the refusal must still be logged once (at V(1)), got %v", *refusals)
+	}
+}
+
+// The recorder must not claim to flush a writer that cannot.
+func TestStatusRecorderFlushFollowsTheWrappedWriter(t *testing.T) {
+	type plainWriter struct{ http.ResponseWriter }
+	sw := &statusRecorder{ResponseWriter: plainWriter{httptest.NewRecorder()}}
+	if err := http.NewResponseController(sw).Flush(); !errors.Is(err, http.ErrNotSupported) {
+		t.Fatalf("Flush through a recorder over a non-flushing writer: %v, want ErrNotSupported", err)
+	}
+	rec := httptest.NewRecorder()
+	if err := http.NewResponseController(&statusRecorder{ResponseWriter: rec}).Flush(); err != nil || !rec.Flushed {
+		t.Fatalf("Flush through a recorder over a flushing writer: err=%v flushed=%v", err, rec.Flushed) // positive control
+	}
+}

weed/s3api/s3tables/handler_namespace.go

@@ -36,10 +36,9 @@ func (h *S3TablesHandler) mayRepairOwnerlessBucket(r *http.Request) bool {
 
 // handleCreateNamespace creates a new namespace in a table bucket
 func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error {
-	glog.Errorf("S3Tables: handleCreateNamespace called")
+	glog.V(3).Infof("S3Tables: handleCreateNamespace called")
 	var req CreateNamespaceRequest
 	if err := h.readRequestBody(r, &req); err != nil {
-		glog.Errorf("S3Tables: handleCreateNamespace failed to read request body: %v", err)
 		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
 		return err
 	}
@@ -140,6 +139,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 +484,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_no_trace_errorf_test.go

@@ -0,0 +1,39 @@
+package s3tables
+
+import (
+	"os"
+	"path/filepath"
+	"regexp"
+	"strings"
+	"testing"
+)
+
+// glog.Errorf with a constant message and no format verb is a trace, not an
+// error report: every CreateNamespace call logged
+// "S3Tables: handleCreateNamespace called" at ERROR.
+func TestNoTraceLoggedAtErrorLevel(t *testing.T) {
+	files, err := filepath.Glob("*.go")
+	if err != nil || len(files) == 0 {
+		t.Fatalf("no sources found (%v)", err) // positive control: the scan must see the package
+	}
+	re := regexp.MustCompile(`glog\.Errorf\("([^"]*)"\)`)
+	scanned := 0
+	for _, f := range files {
+		if strings.HasSuffix(f, "_test.go") {
+			continue
+		}
+		b, err := os.ReadFile(f)
+		if err != nil {
+			t.Fatal(err)
+		}
+		scanned++
+		for _, m := range re.FindAllStringSubmatch(string(b), -1) {
+			if !strings.Contains(m[1], "%") {
+				t.Errorf("%s: glog.Errorf(%q) logs a constant trace at ERROR; use glog.V(n).Infof", f, m[1])
+			}
+		}
+	}
+	if scanned < 5 {
+		t.Fatalf("scanned only %d source files", scanned)
+	}
+}

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/s3api/table_bucket_verify_object_test.go

@@ -0,0 +1,163 @@
+package s3api
+
+import (
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/gorilla/mux"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/stretchr/testify/assert"
+)
+
+const verifyTestBucket = "tables"
+
+// verifyTestNow is the server clock the tests pin; keys are dated around it.
+const verifyTestNow = 1790000000000
+
+func newVerifyTestServer(t *testing.T) *S3ApiServer {
+	t.Helper()
+	saved := volumeVerifyNow
+	volumeVerifyNow = func() time.Time { return time.UnixMilli(verifyTestNow) }
+	savedLookup := volumeVerifyLookup
+	volumeVerifyLookup = func(*S3ApiServer, string, string) (*filer_pb.Entry, error) { return nil, filer_pb.ErrNotFound }
+	t.Cleanup(func() { volumeVerifyNow, volumeVerifyLookup = saved, savedLookup })
+	s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}}
+	s3a.bucketRegistry = NewBucketRegistry(s3a)
+	s3a.bucketRegistry.setMetadataCache(&BucketMetaData{Name: verifyTestBucket, IsTableBucket: true})
+	s3a.bucketRegistry.setMetadataCache(&BucketMetaData{Name: "plain", IsTableBucket: false})
+	return s3a
+}
+
+func TestTableBucketAcceptsExternalVolumeVerifyKey(t *testing.T) {
+	s3a := newVerifyTestServer(t)
+	for _, key := range []string{
+		"verify_1790000000000_1081846483",
+		"/verify_1790000000000_1",
+		"verify_1790003300000_42", // 55 minutes later: inside the window
+		"verify_1789996700000_42", // 55 minutes earlier
+		// reads and deletes take a key of any date (PUT limits it)
+		"verify_1789992000000_1",
+		"verify_1790008000000_1",
+	} {
+		isVerify, err := s3a.validateTableBucketObjectPathOrVerify(verifyTestBucket, key)
+		assert.True(t, isVerify, key)
+		assert.NoError(t, err, key)
+		// the handlers that keep the plain rule still refuse it
+		assert.Error(t, s3a.validateTableBucketObjectPath(verifyTestBucket, key), key)
+	}
+	// the Iceberg layout is unchanged: the positive control
+	isVerify, err := s3a.validateTableBucketObjectPathOrVerify(verifyTestBucket, "ns/t/data/part-00000.parquet")
+	assert.False(t, isVerify)
+	assert.NoError(t, err)
+}
+
+func TestTableBucketRefusesVerifyKeyLookalikes(t *testing.T) {
+	s3a := newVerifyTestServer(t)
+	for _, key := range []string{
+		// the bucket root only
+		"base/verify_1790000000000_1",
+		"ns/verify_1790000000000_1",
+		".uploads/verify_1790000000000_1",
+		".uploads/abc/verify_1790000000000_1",
+		// the exact shape only
+		"verify_1790000000000_1.txt",
+		"verify_1790000000000",
+		"verify_179000000000_1",            // 12-digit timestamp
+		"verify_1790000000000_12345678901", // 11-digit number
+		"verify_abc_123",
+		"verify_1790000000000_1/x",
+		"xverify_1790000000000_1",
+		// ordinary non-Iceberg keys
+		"notes.txt",
+		"ns/t/data/notes.txt",
+	} {
+		isVerify, err := s3a.validateTableBucketObjectPathOrVerify(verifyTestBucket, key)
+		assert.False(t, isVerify, key)
+		assert.Error(t, err, key)
+	}
+	// a bucket that is not a table bucket is unaffected
+	isVerify, err := s3a.validateTableBucketObjectPathOrVerify("plain", "verify_1790000000000_1")
+	assert.False(t, isVerify)
+	assert.NoError(t, err)
+}
+
+func verifyPutRequest(key string) *http.Request {
+	req := httptest.NewRequest(http.MethodPut, "/"+verifyTestBucket+"/"+key, strings.NewReader("x"))
+	return mux.SetURLVars(req, map[string]string{"bucket": verifyTestBucket, "object": key})
+}
+
+// A PUT to the verification key cannot store data: a small declared body and
+// nothing stored beside it.
+func TestTableBucketVerifyKeyPutLimits(t *testing.T) {
+	s3a := newVerifyTestServer(t)
+	const key = "verify_1790000000000_1081846483"
+	for name, mutate := range map[string]func(*http.Request){
+		"too large":                    func(r *http.Request) { r.ContentLength = volumeVerifyMaxBytes + 1 },
+		"content-language":             func(r *http.Request) { r.ContentLength = 0; r.Header.Set("Content-Language", "fr") },
+		"content-encoding":             func(r *http.Request) { r.ContentLength = 0; r.Header.Set("Content-Encoding", "gzip") },
+		"response-content-disposition": func(r *http.Request) { r.ContentLength = 0; r.Header.Set("Response-Content-Disposition", "attachment") },
+		"extract query":                func(r *http.Request) { r.ContentLength = 0; r.URL.RawQuery = "extract=zip" },
+		"extract header":               func(r *http.Request) { r.ContentLength = 0; r.Header.Set(s3_constants.SeaweedFSExtract, "zip") },
+		"undeclared length":            func(r *http.Request) { r.ContentLength = -1 },
+		"chunked":                      func(r *http.Request) { r.ContentLength = 0; r.TransferEncoding = []string{"chunked"} },
+		"user metadata":                func(r *http.Request) { r.ContentLength = 0; r.Header.Set("X-Amz-Meta-Note", "x") },
+		"tagging":                      func(r *http.Request) { r.ContentLength = 0; r.Header.Set("X-Amz-Tagging", "a=b") },
+		"content-disposition":          func(r *http.Request) { r.ContentLength = 0; r.Header.Set("Content-Disposition", "attachment") },
+		"website redirect":             func(r *http.Request) { r.ContentLength = 0; r.Header.Set("X-Amz-Website-Redirect-Location", "/x") },
+	} {
+		req := verifyPutRequest(key)
+		mutate(req)
+		assert.False(t, volumeVerifyPutAllowed(req, key), name)
+		rec := httptest.NewRecorder()
+		s3a.PutObjectHandler(rec, req)
+		assert.Equal(t, http.StatusForbidden, rec.Code, name)
+	}
+	ok := verifyPutRequest(key)
+	ok.ContentLength = 21
+	ok.Header.Set("Content-Type", "application/octet-stream")
+	ok.Header.Set("Content-Encoding", "aws-chunked")
+	assert.True(t, volumeVerifyPutAllowed(ok, key), "the verification object itself")
+
+	// a PUT is dated within the window of the server's clock
+	for _, old := range []string{"verify_1789992000000_1", "verify_1790008000000_1"} {
+		req := verifyPutRequest(old)
+		req.ContentLength = 21
+		assert.False(t, volumeVerifyPutAllowed(req, old), old)
+		rec := httptest.NewRecorder()
+		s3a.PutObjectHandler(rec, req)
+		assert.Equal(t, http.StatusForbidden, rec.Code, old)
+	}
+}
+
+// A namespace may carry a name of the key's shape; the key then names its
+// directory and gets the plain layout rule, as does a key whose entry cannot
+// be read.
+func TestTableBucketVerifyKeyNamingDirectoryIsRefused(t *testing.T) {
+	s3a := newVerifyTestServer(t)
+	const key = "verify_1790000000000_7"
+	for name, lookup := range map[string]func(*S3ApiServer, string, string) (*filer_pb.Entry, error){
+		"namespace directory": func(*S3ApiServer, string, string) (*filer_pb.Entry, error) {
+			return &filer_pb.Entry{Name: key, IsDirectory: true}, nil
+		},
+		"lookup failure": func(*S3ApiServer, string, string) (*filer_pb.Entry, error) {
+			return nil, errors.New("filer unavailable")
+		},
+	} {
+		volumeVerifyLookup = lookup
+		isVerify, err := s3a.validateTableBucketObjectPathOrVerify(verifyTestBucket, key)
+		assert.False(t, isVerify, name)
+		assert.Error(t, err, name)
+	}
+	// the control: an existing verification object is a file
+	volumeVerifyLookup = func(*S3ApiServer, string, string) (*filer_pb.Entry, error) {
+		return &filer_pb.Entry{Name: key}, nil
+	}
+	isVerify, err := s3a.validateTableBucketObjectPathOrVerify(verifyTestBucket, key)
+	assert.True(t, isVerify)
+	assert.NoError(t, 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_grpc_tier_upload.go

@@ -110,6 +110,8 @@ func (vs *VolumeServer) VolumeTierMoveDatToRemote(req *volume_server_pb.VolumeTi
 	if err := v.LoadRemoteFile(); err != nil {
 		return fmt.Errorf("volume %d failed to load remote file: %v", v.Id, err)
 	}
+	// Deletes must reach the index now, not only after the next restart.
+	v.MarkTieredDeletable()
 
 	if !req.KeepLocalDatFile {
 		os.Remove(v.FileName(".dat"))

weed/server/volume_grpc_tier_upload_deletes_test.go

@@ -0,0 +1,77 @@
+package weed_server
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
+	"github.com/seaweedfs/seaweedfs/weed/storage"
+	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// tierUploadForDeletes writes 10 needles into volume 1 of a new store, marks it
+// read-only as volume.tier.upload does, and moves its .dat to the tier.
+func tierUploadForDeletes(t *testing.T, replication string) *storage.Store {
+	t.Helper()
+	backend.BackendStorages[tierTimestampTestBackendName] = &tierTimestampTestBackend{root: t.TempDir()}
+	t.Cleanup(func() { delete(backend.BackendStorages, tierTimestampTestBackendName) })
+	store := storage.NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
+		[]string{t.TempDir()}, []int32{10}, []util.MinFreeSpace{{}}, "", storage.NeedleMapInMemory,
+		[]types.DiskType{types.HardDriveType}, nil, 0, stats.DefaultDiskIOProbeConfig())
+	t.Cleanup(store.Close)
+	if err := store.AddVolume(1, "", storage.NeedleMapInMemory, replication, "", 0, needle.Version3, 0, types.HardDriveType, 0); err != nil {
+		t.Fatalf("add volume: %v", err)
+	}
+	for i := 1; i <= 10; i++ {
+		n := &needle.Needle{Id: types.Uint64ToNeedleId(uint64(i)), Cookie: 7, Data: bytes.Repeat([]byte{byte(i)}, 1024)}
+		n.Checksum = needle.NewCRC(n.Data)
+		if _, err := store.WriteVolumeNeedle(1, n, false, false); err != nil {
+			t.Fatalf("write %d: %v", i, err)
+		}
+	}
+	if err := store.MarkVolumeReadonly(1, false, false); err != nil {
+		t.Fatalf("mark read-only: %v", err)
+	}
+	vs := &VolumeServer{store: store}
+	if err := vs.VolumeTierMoveDatToRemote(&volume_server_pb.VolumeTierMoveDatToRemoteRequest{
+		VolumeId: 1, DestinationBackendName: tierTimestampTestBackendName,
+	}, &discardServerStream[volume_server_pb.VolumeTierMoveDatToRemoteResponse]{}); err != nil {
+		t.Fatalf("tier upload: %v", err)
+	}
+	return store
+}
+
+// A single-copy volume takes deletes as soon as it is tiered, as it would
+// after a restart. Before, they were refused ("volume 1 is read only") until
+// then, leaving the needles live in the index.
+func TestTierUploadAcceptsDeletesOnASingleCopyVolume(t *testing.T) {
+	store := tierUploadForDeletes(t, "000")
+	if _, err := store.DeleteVolumeNeedle(1, &needle.Needle{Id: 2}); err != nil {
+		t.Fatalf("delete on the freshly tiered volume: %v", err)
+	}
+	if _, err := store.ReadVolumeNeedle(1, &needle.Needle{Id: 2}, nil, nil); err == nil {
+		t.Fatalf("the deleted needle still reads")
+	}
+	if _, err := store.ReadVolumeNeedle(1, &needle.Needle{Id: 3}, nil, nil); err != nil {
+		t.Fatalf("a live needle does not read: %v", err)
+	}
+	n := &needle.Needle{Id: 99, Cookie: 7, Data: []byte("x")}
+	n.Checksum = needle.NewCRC(n.Data)
+	if _, err := store.WriteVolumeNeedle(1, n, false, false); err == nil {
+		t.Fatalf("a write to a tiered volume succeeded")
+	}
+}
+
+// A replicated volume keeps refusing deletes: the shell copies the uploaded
+// replica's .idx and .vif to the others after the upload, and a tombstone
+// written before that copy would reach one replica only.
+func TestTierUploadKeepsRefusingDeletesOnAReplicatedVolume(t *testing.T) {
+	store := tierUploadForDeletes(t, "001")
+	if _, err := store.DeleteVolumeNeedle(1, &needle.Needle{Id: 2}); err == nil {
+		t.Fatalf("a replicated volume accepted a delete right after the tier upload")
+	}
+}

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/backend/s3_backend/s3_backend.go

@@ -3,6 +3,7 @@ package s3_backend
 import (
 	"fmt"
 	"io"
+	"math"
 	"os"
 	"strconv"
 	"strings"
@@ -44,6 +45,7 @@ type S3BackendStorage struct {
 	forcePathStyle        bool
 	uploadConcurrency     int
 	downloadConcurrency   int
+	httpOptions           httpOptions
 	conn                  s3iface.S3API
 }
 
@@ -59,11 +61,16 @@ func newS3BackendStorage(configuration backend.StringProperties, configPrefix st
 	s.forcePathStyle = util.ParseBool(configuration.GetString(configPrefix+"force_path_style"), true)
 	s.uploadConcurrency = parseConcurrency(configuration.GetString(configPrefix+"upload_concurrency"), defaultUploadConcurrency)
 	s.downloadConcurrency = parseConcurrency(configuration.GetString(configPrefix+"download_concurrency"), defaultDownloadConcurrency)
+	s.httpOptions = httpOptions{
+		disableHTTP2:    util.ParseBool(configuration.GetString(configPrefix+"disable_http2"), false),
+		maxConnsPerHost: parseNonNegative(configuration.GetString(configPrefix+"max_connections"), "max_connections"),
+		idleConnTimeout: parseIdleConnTimeout(configuration.GetString(configPrefix + "idle_connection_timeout_seconds")),
+	}
 	if s.storageClass == "" {
 		s.storageClass = "STANDARD_IA"
 	}
 
-	s.conn, err = createSession(s.aws_access_key_id, s.aws_secret_access_key, s.region, s.endpoint, s.forcePathStyle)
+	s.conn, err = createSession(s.aws_access_key_id, s.aws_secret_access_key, s.region, s.endpoint, s.forcePathStyle, s.httpOptions)
 
 	glog.V(0).Infof("created backend storage s3.%s for region %s bucket %s", s.id, s.region, s.bucket)
 	return
@@ -80,6 +87,9 @@ func (s *S3BackendStorage) ToProperties() map[string]string {
 	m["force_path_style"] = util.BoolToString(s.forcePathStyle)
 	m["upload_concurrency"] = strconv.Itoa(s.uploadConcurrency)
 	m["download_concurrency"] = strconv.Itoa(s.downloadConcurrency)
+	m["disable_http2"] = util.BoolToString(s.httpOptions.disableHTTP2)
+	m["max_connections"] = strconv.Itoa(s.httpOptions.maxConnsPerHost)
+	m["idle_connection_timeout_seconds"] = strconv.FormatInt(int64(s.httpOptions.idleConnTimeout/time.Second), 10)
 	return m
 }
 
@@ -99,6 +109,42 @@ func parseConcurrency(value string, def int) int {
 	return def
 }
 
+// parseNonNegative reads an optional non-negative integer setting; empty or
+// invalid means 0 (the default).
+func parseNonNegative(value, name string) int {
+	if value == "" {
+		return 0
+	}
+	if n, err := strconv.Atoi(value); err == nil && n >= 0 {
+		return n
+	}
+	glog.Warningf("invalid %s value %q, using the default", name, value)
+	return 0
+}
+
+// maxIdleConnTimeoutSeconds is the largest number of seconds a time.Duration
+// can hold.
+const maxIdleConnTimeoutSeconds = math.MaxInt64 / int64(time.Second)
+
+// parseIdleConnTimeout reads idle_connection_timeout_seconds; empty, invalid
+// or too large to represent as a time.Duration means 0 (the default). It
+// parses 64-bit seconds so the range is the same on 32-bit platforms.
+func parseIdleConnTimeout(value string) time.Duration {
+	if value == "" {
+		return 0
+	}
+	seconds, err := strconv.ParseInt(value, 10, 64)
+	if err != nil || seconds < 0 {
+		glog.Warningf("invalid idle_connection_timeout_seconds value %q, using the default", value)
+		return 0
+	}
+	if seconds > maxIdleConnTimeoutSeconds {
+		glog.Warningf("idle_connection_timeout_seconds %q is out of range, using the default", value)
+		return 0
+	}
+	return time.Duration(seconds) * time.Second
+}
+
 func (s *S3BackendStorage) NewStorageFile(key string, tierInfo *volume_server_pb.VolumeInfo) backend.BackendStorageFile {
 	if strings.HasPrefix(key, "/") {
 		key = key[1:]

weed/storage/backend/s3_backend/s3_sessions.go

@@ -1,8 +1,11 @@
 package s3_backend
 
 import (
+	"crypto/tls"
 	"fmt"
+	"net/http"
 	"sync"
+	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/credentials"
@@ -18,12 +21,49 @@ var (
 	sessionsLock sync.Mutex
 )
 
-func createSession(awsAccessKeyId, awsSecretAccessKey, region, endpoint string, forcePathStyle bool) (s3iface.S3API, error) {
+// httpOptions tunes the HTTP transport of the tier backend's S3 client. The
+// zero value keeps the SDK's default client.
+type httpOptions struct {
+	// disableHTTP2 keeps the client on HTTP/1.1. Over HTTP/2 every upload and
+	// range read to the endpoint shares one connection and its flow-control
+	// window, which some S3-compatible services keep small.
+	disableHTTP2 bool
+	// maxConnsPerHost caps the connections to the endpoint (0 = no cap), for
+	// services that limit parallel connections per client address.
+	maxConnsPerHost int
+	// idleConnTimeout closes idle connections sooner than Go's default 90s,
+	// for services that drop idle connections earlier (0 = default).
+	idleConnTimeout time.Duration
+}
+
+// newTransport returns the transport for the options, or nil for the default.
+func (o httpOptions) newTransport() *http.Transport {
+	if o == (httpOptions{}) {
+		return nil
+	}
+	t := http.DefaultTransport.(*http.Transport).Clone()
+	if o.disableHTTP2 {
+		t.ForceAttemptHTTP2 = false
+		t.TLSNextProto = map[string]func(string, *tls.Conn) http.RoundTripper{}
+	}
+	if o.maxConnsPerHost > 0 {
+		t.MaxConnsPerHost = o.maxConnsPerHost
+	}
+	if o.idleConnTimeout > 0 {
+		t.IdleConnTimeout = o.idleConnTimeout
+	}
+	return t
+}
+
+func createSession(awsAccessKeyId, awsSecretAccessKey, region, endpoint string, forcePathStyle bool, httpOpts httpOptions) (s3iface.S3API, error) {
 
 	sessionsLock.Lock()
 	defer sessionsLock.Unlock()
 
 	cacheKey := fmt.Sprintf("%s|%s", region, endpoint)
+	if httpOpts != (httpOptions{}) {
+		cacheKey = fmt.Sprintf("%s|%+v", cacheKey, httpOpts)
+	}
 	if t, found := s3Sessions[cacheKey]; found {
 		return t, nil
 	}
@@ -34,6 +74,9 @@ func createSession(awsAccessKeyId, awsSecretAccessKey, region, endpoint string,
 		S3ForcePathStyle:              aws.Bool(forcePathStyle),
 		S3DisableContentMD5Validation: aws.Bool(true),
 	}
+	if t := httpOpts.newTransport(); t != nil {
+		config.HTTPClient = &http.Client{Transport: t}
+	}
 	if awsAccessKeyId != "" && awsSecretAccessKey != "" {
 		config.Credentials = credentials.NewStaticCredentials(awsAccessKeyId, awsSecretAccessKey, "")
 	}

weed/storage/backend/s3_backend/s3_sessions_test.go

@@ -0,0 +1,129 @@
+package s3_backend
+
+import (
+	"math"
+	"net/http"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/service/s3"
+)
+
+func backendTransport(t *testing.T, s *S3BackendStorage) *http.Transport {
+	t.Helper()
+	client, ok := s.conn.(*s3.S3)
+	if !ok {
+		t.Fatalf("unexpected client type %T", s.conn)
+	}
+	httpClient := client.Client.Config.HTTPClient
+	if httpClient == nil {
+		return nil
+	}
+	transport, _ := httpClient.Transport.(*http.Transport)
+	return transport
+}
+
+func TestS3BackendHTTPOptionsReachTheClient(t *testing.T) {
+	s, err := newS3BackendStorage(testProperties{
+		"endpoint":                        "http://127.0.0.1:1",
+		"disable_http2":                   "true",
+		"max_connections":                 "64",
+		"idle_connection_timeout_seconds": "50",
+	}, "", "http-options")
+	if err != nil {
+		t.Fatal(err)
+	}
+	transport := backendTransport(t, s)
+	if transport == nil {
+		t.Fatal("configured HTTP options did not reach the S3 client")
+	}
+	if transport.ForceAttemptHTTP2 || transport.TLSNextProto == nil || len(transport.TLSNextProto) != 0 {
+		t.Errorf("HTTP/2 not disabled: ForceAttemptHTTP2=%v TLSNextProto=%v", transport.ForceAttemptHTTP2, transport.TLSNextProto)
+	}
+	if transport.MaxConnsPerHost != 64 {
+		t.Errorf("MaxConnsPerHost = %d, want 64", transport.MaxConnsPerHost)
+	}
+	if transport.IdleConnTimeout != 50*time.Second {
+		t.Errorf("IdleConnTimeout = %v, want 50s", transport.IdleConnTimeout)
+	}
+
+	props := s.ToProperties()
+	if props["disable_http2"] != "true" || props["max_connections"] != "64" || props["idle_connection_timeout_seconds"] != "50" {
+		t.Errorf("ToProperties: %v", props)
+	}
+}
+
+func TestS3BackendDefaultHTTPClientUnchanged(t *testing.T) {
+	s, err := newS3BackendStorage(testProperties{"endpoint": "http://127.0.0.1:2"}, "", "http-defaults")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if transport := backendTransport(t, s); transport != nil && transport != http.DefaultTransport {
+		t.Fatalf("no options set, yet the client got a custom transport %+v", transport)
+	}
+	if (httpOptions{}).newTransport() != nil {
+		t.Fatal("zero options must keep the default client")
+	}
+}
+
+func TestS3BackendSessionsDifferByHTTPOptions(t *testing.T) {
+	a, err := createSession("", "", "us-east-1", "http://127.0.0.1:3", true, httpOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	b, err := createSession("", "", "us-east-1", "http://127.0.0.1:3", true, httpOptions{disableHTTP2: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if a == b {
+		t.Fatal("backends with different HTTP options share one client")
+	}
+	c, err := createSession("", "", "us-east-1", "http://127.0.0.1:3", true, httpOptions{disableHTTP2: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if b != c {
+		t.Fatal("the same endpoint and options must reuse the cached client")
+	}
+}
+
+func TestParseNonNegative(t *testing.T) {
+	for _, c := range []struct {
+		in   string
+		want int
+	}{{"", 0}, {"0", 0}, {"12", 12}, {"-1", 0}, {"x", 0}} {
+		if got := parseNonNegative(c.in, "test"); got != c.want {
+			t.Errorf("parseNonNegative(%q) = %d, want %d", c.in, got, c.want)
+		}
+	}
+}
+
+func TestS3BackendIdleTimeoutOverflowFallsBackToDefault(t *testing.T) {
+	maxSeconds := int64(math.MaxInt64 / int64(time.Second))
+	for i, c := range []struct {
+		in   string
+		want time.Duration
+	}{
+		{strconv.FormatInt(maxSeconds, 10), time.Duration(maxSeconds) * time.Second},
+		{strconv.FormatInt(maxSeconds+1, 10), 0},
+		{"10000000000", 0},
+		{"-1", 0},
+		{"x", 0},
+	} {
+		s, err := newS3BackendStorage(testProperties{
+			"endpoint":                        "http://127.0.0.1:4",
+			"idle_connection_timeout_seconds": c.in,
+		}, "", "idle-overflow-"+strconv.Itoa(i))
+		if err != nil {
+			t.Fatal(err)
+		}
+		if s.httpOptions.idleConnTimeout != c.want {
+			t.Errorf("idle_connection_timeout_seconds=%s: timeout %v, want %v", c.in, s.httpOptions.idleConnTimeout, c.want)
+		}
+		want := strconv.FormatInt(int64(c.want/time.Second), 10)
+		if got := s.ToProperties()["idle_connection_timeout_seconds"]; got != want {
+			t.Errorf("idle_connection_timeout_seconds=%s: ToProperties exports %s, want %s", c.in, got, want)
+		}
+	}
+}

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
@@ -869,7 +880,10 @@ func (s *Store) WriteVolumeNeedle(i needle.VolumeId, n *needle.Needle, checkCook
 
 func (s *Store) DeleteVolumeNeedle(i needle.VolumeId, n *needle.Needle) (Size, error) {
 	if v := s.findVolume(i); v != nil {
-		if v.noWriteOrDelete {
+		v.noWriteLock.RLock()
+		noWriteOrDelete := v.noWriteOrDelete
+		v.noWriteLock.RUnlock()
+		if noWriteOrDelete {
 			return 0, fmt.Errorf("volume %d is read only", i)
 		}
 		return v.deleteNeedle2(n)
@@ -975,10 +989,10 @@ func (s *Store) MarkVolumeWritable(i needle.VolumeId) error {
 	prevNoWriteOrDelete := v.noWriteOrDelete
 	prevNoWriteCanDelete := v.noWriteCanDelete
 	v.noWriteOrDelete = false
-	// Remote-tiered volumes must stay noWriteCanDelete regardless of marks.
-	if !v.HasRemoteFile() {
-		v.noWriteCanDelete = false
-	}
+	// A remote-tiered volume keeps the delete-only mode load() derives: its
+	// .dat cannot take appends, so "writable" resolves to can-delete rather
+	// than advertising writable with no local data file.
+	v.noWriteCanDelete = v.HasRemoteFile()
 	persistErr := v.PersistReadOnly(false, false)
 	if persistErr != nil {
 		var ndErr *volume_info.NotCrashDurableError
@@ -1302,11 +1316,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)

weed/storage/volume_loading.go

@@ -238,7 +238,7 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 		glog.V(2).Infof("readSuperBlock volume %d version %v", v.Id, v.SuperBlock.Version)
 		if v.HasRemoteFile() {
 			// maybe temporary network problem
-			glog.Errorf("readSuperBlock remote volume %d: %v", v.Id, err)
+			logRemoteSuperBlockErr(v.Id, err)
 			err = nil
 		}
 	} else {
@@ -434,3 +434,11 @@ func datFileLoadError(fileName string, err error) error {
 	}
 	return fmt.Errorf("cannot load volume data %s: %v", fileName, err)
 }
+
+var logRemoteSuperBlockFault = glog.Errorf
+
+func logRemoteSuperBlockErr(id needle.VolumeId, err error) {
+	if err != nil {
+		logRemoteSuperBlockFault("readSuperBlock remote volume %d: %v", id, err)
+	}
+}

weed/storage/volume_mark_writable_test.go

@@ -68,6 +68,33 @@ func TestMarkVolumeWritable_ReopensPersistedReadOnly(t *testing.T) {
 	}
 }
 
+// A remote-tiered volume marked writable keeps the delete-only mode load()
+// derives — the remote .dat cannot take appends, so "writable" resolves to
+// can-delete even when the mark left noWriteOrDelete without noWriteCanDelete.
+func TestMarkVolumeWritableOnATieredVolumeKeepsDeletesOnly(t *testing.T) {
+	b := newLocalDirBackend(t)
+	registerTestBackend(t, b)
+
+	v, _ := tierUpVolumeLive(t, t.TempDir(), 86, b)
+	defer v.Close()
+	// the upload-time mark as a refused MarkTieredDeletable leaves it
+	v.noWriteOrDelete = true
+	v.noWriteCanDelete = false
+
+	location := newTestLocation(10, false, 0)
+	location.SetVolume(86, v)
+	s := &Store{Locations: []*DiskLocation{location}}
+	if err := s.MarkVolumeWritable(86); err != nil {
+		t.Fatalf("mark writable: %v", err)
+	}
+
+	readOnly, orDelete, canDelete, _ := v.ReadOnlyReasons()
+	if orDelete || !canDelete || !readOnly {
+		t.Fatalf("a tiered volume resolves writable to delete-only, got readOnly=%v orDelete=%v canDelete=%v",
+			readOnly, orDelete, canDelete)
+	}
+}
+
 // reopenIdxForWrite must be a no-op when the volume already has a writable
 // needle map — otherwise repeated MarkVolumeWritable calls would churn the
 // index file handle for no reason.

weed/storage/volume_remote_superblock_log_test.go

@@ -0,0 +1,21 @@
+package storage
+
+import (
+	"errors"
+	"testing"
+)
+
+func TestRemoteSuperBlockLogOnlyOnError(t *testing.T) {
+	var logged int
+	old := logRemoteSuperBlockFault
+	logRemoteSuperBlockFault = func(string, ...interface{}) { logged++ }
+	defer func() { logRemoteSuperBlockFault = old }()
+	logRemoteSuperBlockErr(219, nil)
+	if logged != 0 {
+		t.Fatalf("a nil error was logged at ERROR (the \"readSuperBlock remote volume 219: <nil>\" line)")
+	}
+	logRemoteSuperBlockErr(219, errors.New("timeout")) // positive control
+	if logged != 1 {
+		t.Fatalf("a real error must still be logged, got %d", logged)
+	}
+}

weed/storage/volume_tier.go

@@ -87,6 +87,41 @@ func (v *Volume) LoadRemoteFile() error {
 	return v.loadRemoteFileLocked()
 }
 
+// MarkTieredDeletable gives a volume whose data has just moved to the remote
+// tier the flags a restart gives it (load(): a tiered volume refuses writes and
+// accepts deletes). volume.tier.upload marks the volume read-only without
+// deletes before the upload; left that way, every delete until the next
+// restart is refused, the filer drops it, and the needle's bytes stay in the
+// remote object where no vacuum can find them.
+//
+// Only for a single-copy volume: the shell copies the uploaded replica's .idx
+// and .vif to the other replicas after the upload, and a tombstone written
+// before that copy would reach one replica only; a replicated volume keeps
+// refusing deletes until its next load, as before. An I/O quarantine is left
+// alone, and so is a volume past the I/O error tolerance that the next
+// heartbeat will quarantine. Otherwise the flags are the ones load() gives
+// every tiered volume.
+func (v *Volume) MarkTieredDeletable() bool {
+	if rp := v.SuperBlock.ReplicaPlacement; rp != nil && rp.GetCopyCount() > 1 {
+		return false
+	}
+	// Check the I/O state under noWriteLock: CollectHeartbeat marks the
+	// quarantine before it takes noWriteLock to refuse deletes, so either the
+	// check below sees the quarantine, or the heartbeat's refusal lands after
+	// this change and stays.
+	v.noWriteLock.Lock()
+	defer v.noWriteLock.Unlock()
+	if !v.HasRemoteFile() {
+		return false
+	}
+	if ioErr, ioCount, quarantined := v.getIoErrorState(); quarantined || (ioErr != nil && ioCount >= IoErrorTolerance) {
+		return false
+	}
+	v.noWriteOrDelete = false
+	v.noWriteCanDelete = true
+	return true
+}
+
 // loadRemoteFileLocked swaps the data backend to the remote tier object. The
 // caller must hold dataFileAccessLock or be single-threaded (load() during
 // construction or a compaction-commit reload). It marks the volume tiered in the

weed/storage/volume_tier_deletable_test.go

@@ -0,0 +1,123 @@
+package storage
+
+import (
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/stretchr/testify/require"
+)
+
+// MarkTieredDeletable gives a freshly tiered single-copy volume the flags a
+// restart gives it, and leaves an I/O quarantine alone.
+func TestMarkTieredDeletable(t *testing.T) {
+	b := newLocalDirBackend(t)
+	registerTestBackend(t, b)
+
+	v, _ := tierUpVolumeLive(t, t.TempDir(), 81, b)
+	defer v.Close()
+	v.noWriteOrDelete = true // as volume.tier.upload's mark leaves it
+	require.True(t, v.MarkTieredDeletable())
+	require.False(t, v.noWriteOrDelete)
+	require.True(t, v.noWriteCanDelete)
+
+	w, _ := tierUpVolumeLive(t, t.TempDir(), 82, b)
+	defer w.Close()
+	w.noWriteOrDelete = true
+	w.markIoQuarantined()
+	require.False(t, w.MarkTieredDeletable())
+	require.True(t, w.noWriteOrDelete, "an I/O quarantine is not the tier upload's to lift")
+}
+
+// MarkTieredDeletable is only for a tiered volume: a stray call on a local
+// volume must not pin it read-only-can-delete.
+func TestMarkTieredDeletableLeavesANonTieredVolume(t *testing.T) {
+	dir := t.TempDir()
+	v, err := NewVolume(dir, dir, "", 86, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	require.NoError(t, err)
+	defer v.Close()
+
+	require.False(t, v.MarkTieredDeletable())
+	require.False(t, v.noWriteOrDelete)
+	require.False(t, v.noWriteCanDelete)
+}
+
+// A heartbeat that quarantines the volume while MarkTieredDeletable waits for
+// noWriteLock must keep its refusal: the quarantine check and the flag change
+// happen under the same lock.
+func TestMarkTieredDeletableKeepsAConcurrentQuarantine(t *testing.T) {
+	b := newLocalDirBackend(t)
+	registerTestBackend(t, b)
+
+	v, _ := tierUpVolumeLive(t, t.TempDir(), 83, b)
+	defer v.Close()
+	v.noWriteOrDelete = true
+
+	v.noWriteLock.Lock()
+	done := make(chan bool)
+	go func() { done <- v.MarkTieredDeletable() }()
+	time.Sleep(100 * time.Millisecond) // let it reach noWriteLock
+	// What CollectHeartbeat does: mark the quarantine, then refuse writes and deletes.
+	v.markIoQuarantined()
+	v.noWriteOrDelete = true
+	v.noWriteLock.Unlock()
+
+	require.False(t, <-done, "a volume quarantined before the flags changed must not be made deletable")
+	v.noWriteLock.RLock()
+	defer v.noWriteLock.RUnlock()
+	require.True(t, v.noWriteOrDelete, "the heartbeat's refusal was overwritten")
+}
+
+// A volume past the I/O error tolerance is about to be quarantined by the next
+// heartbeat; it is not made deletable in the meantime.
+func TestMarkTieredDeletableRefusesAVolumePastIoErrorTolerance(t *testing.T) {
+	b := newLocalDirBackend(t)
+	registerTestBackend(t, b)
+
+	v, _ := tierUpVolumeLive(t, t.TempDir(), 84, b)
+	defer v.Close()
+	v.noWriteOrDelete = true
+	for i := 0; i < IoErrorTolerance; i++ {
+		v.noteIoError(syscall.EIO)
+	}
+	require.False(t, v.MarkTieredDeletable())
+	require.True(t, v.noWriteOrDelete)
+
+	v.clearIoError() // positive control: below the tolerance it is made deletable
+	require.True(t, v.MarkTieredDeletable())
+	require.False(t, v.noWriteOrDelete)
+}
+
+// DeleteVolumeNeedle reads the read-only flags under noWriteLock, so a delete
+// that races the tier upload's flag change is not a data race (run with -race).
+func TestDeleteVolumeNeedleRacesMarkTieredDeletable(t *testing.T) {
+	b := newLocalDirBackend(t)
+	registerTestBackend(t, b)
+
+	v, _ := tierUpVolumeLive(t, t.TempDir(), 85, b)
+	defer v.Close()
+	v.noWriteOrDelete = true
+	location := newTestLocation(10, false, 0)
+	location.SetVolume(85, v)
+	s := &Store{Locations: []*DiskLocation{location}}
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for i := 0; i < 100; i++ {
+			v.MarkTieredDeletable()
+		}
+	}()
+	for i := 0; i < 100; i++ {
+		_, _ = s.DeleteVolumeNeedle(85, &needle.Needle{Id: types.Uint64ToNeedleId(1000)})
+	}
+	<-done
+
+	// Positive control: once deletable, a delete of a live needle goes through.
+	size, err := s.DeleteVolumeNeedle(85, &needle.Needle{Id: types.Uint64ToNeedleId(1)})
+	require.NoError(t, err)
+	require.NotZero(t, size)
+}