s3tables: let resource-policy denies veto the learn-absence answer

Review feedback: a caller whose only grant was an identity-level action could still learn absence under a resource-policy Deny, and authorizeAbsentResource evaluated the policy against the caller-supplied ARN, which accepts any account. Denies are now checked before the learn-absence fallback and the policy is evaluated against the canonical ARN.

seaweed-volume/src/server/grpc_server.rs

@@ -53,12 +53,6 @@ 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
@@ -350,46 +344,22 @@ impl VolumeGrpcService {
         is_readonly: bool,
     ) -> Result<(), Status> {
         // VolumeMarkReadonly mutates raft-replicated master topology, so it must
-        // 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;
+        // 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()
+            }
+        };
         if master_url.is_empty() {
-            if !self.state.has_master {
-                return Ok(());
-            }
-            return Err(Status::unavailable("no master leader known"));
+            return Ok(());
         }
-        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| {
+        let grpc_addr = parse_grpc_address(&master_url).map_err(|e| {
             Status::internal(format!("invalid master address {}: {}", master_url, e))
         })?;
         let channel = connect_channel(
@@ -5806,8 +5776,7 @@ impl VolumeCopyJob {
             (&self.dest.data_base_name, self.dest.has_remote_dat);
         let mut preallocate_size: i64 = 0;
         if !has_remote_dat {
-            let master = state.current_master().await;
-            let grpc_addr = super::heartbeat::to_grpc_address(&master);
+            let grpc_addr = super::heartbeat::to_grpc_address(&state.master_url);
             // 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
@@ -5832,7 +5801,7 @@ impl VolumeCopyJob {
                     }
                 }
                 Err(e) => {
-                    tracing::warn!("get master {} configuration: {}", master, e);
+                    tracing::warn!("get master {} configuration: {}", state.master_url, 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.current_master().await,
+                &state.master_url,
                 volume_id,
             )
             .await
@@ -675,7 +675,7 @@ async fn do_replicated_request(
     let locations = lookup_volume(
         &state.http_client,
         &state.outgoing_http_scheme,
-        &state.current_master().await,
+        &state.master_url,
         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.current_master().await,
+        &state.master_url,
         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.current_master().await.is_empty()
+        || state.master_url.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.current_master().await.is_empty() && {
+            !is_proxied_query(&query_string) && !state.master_url.is_empty() && {
                 let store = state.store.read().unwrap();
                 store
                     .find_volume(vid)
@@ -2960,9 +2960,7 @@ pub async fn post_handler(
 
     // Replicate to remote volume servers if this volume has replicas.
     // Matches Go's GetWritableRemoteReplications: skip if copy_count == 1.
-    // 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() {
+    if !is_replicate && write_result.is_ok() && !state.master_url.is_empty() {
         let needs_replication = {
             let store = state.store.read().unwrap();
             store
@@ -3360,7 +3358,7 @@ pub async fn delete_handler(
     };
 
     let is_replicate = del_query.split('&').any(|p| p == "type=replicate");
-    if !is_replicate && delete_result.is_ok() {
+    if !is_replicate && delete_result.is_ok() && !state.master_url.is_empty() {
         let needs_replication = {
             let store = state.store.read().unwrap();
             store
@@ -4234,7 +4232,7 @@ async fn read_remote_chunk_needle(
     let locations = lookup_volume(
         &state.http_client,
         &state.outgoing_http_scheme,
-        &state.current_master().await,
+        &state.master_url,
         vid.0,
     )
     .await?;
@@ -6673,8 +6671,7 @@ 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.clone();
-        s.current_master_url = tokio::sync::RwLock::new(master_url);
+        s.master_url = master_url;
         state
     }
 

seaweed-volume/src/server/heartbeat.rs

@@ -155,11 +155,7 @@ pub async fn run_heartbeat_with_state(
                 }
                 PostAction::SleepPulse => {
                     new_leader = None;
-                    // 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;
+                    tokio::time::sleep(pulse).await;
                 }
             }
 
@@ -258,56 +254,10 @@ 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)
 }
@@ -480,6 +430,14 @@ 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;
@@ -493,31 +451,13 @@ 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)) => {
-                        // 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;
-                        }
+                        // Match Go ordering: DuplicatedUuids first, then volume
+                        // options, then leader redirect.
                         if !hb_resp.duplicated_uuids.is_empty() {
                             let uuids = hb_resp.duplicated_uuids.clone();
                             let duplicate_dirs = off_runtime(config, state, move |_, state| {
@@ -562,6 +502,11 @@ 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,9 +1139,16 @@ async fn cached_lookup_ec_shard_locations(
     state: &Arc<VolumeServerState>,
     vid: VolumeId,
 ) -> io::Result<HashMap<ShardId, Vec<String>>> {
-    let master = state.current_master().await;
+    let master = {
+        let live = state.current_master_url.read().await.clone();
+        if !live.is_empty() {
+            live
+        } else {
+            state.master_url.clone()
+        }
+    };
     if master.is_empty() {
-        return Err(io::Error::other("no master leader known for ec shard lookup"));
+        return Err(io::Error::other("no master configured for ec shard lookup"));
     }
 
     let grpc_addr =

seaweed-volume/src/server/volume_server.rs

@@ -86,8 +86,7 @@ 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 configured master address (e.g., "localhost:9333"). A seed, not a
-    /// lookup target: lookups go to `current_master_url`.
+    /// First master address for volume lookups (e.g., "localhost:9333").
     pub master_url: String,
     /// Seed master addresses for UI rendering.
     pub master_urls: Vec<String>,
@@ -95,12 +94,10 @@ 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>,
-    /// 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
+    /// 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
     /// `currentMasterLock`.
     pub current_master_url: tokio::sync::RwLock<String>,
     /// This server's own address (ip:port) for filtering self from lookup results.
@@ -143,13 +140,6 @@ 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) {
@@ -219,7 +209,7 @@ impl VolumeServerState {
             return false;
         }
         let current = self.current_master_url.read().await.clone();
-        if !current.is_empty() && to_http_address(&current) == key {
+        if !current.is_empty() && current == key {
             return true;
         }
         self.seed_master_set.contains(&key)

seaweed-volume/src/storage/volume.rs

@@ -4185,10 +4185,8 @@ 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();
-        // 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();
+        // 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();
         self.no_write.store(
             if no_write_can_delete {
                 NO_WRITE_CAN_DELETE
@@ -4302,70 +4300,21 @@ impl Volume {
         }
         f(&mut self.volume_info.write().unwrap().files);
         let mut io = self.io.write().unwrap();
-        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;
-            }
-        }
+        self.refresh_remote_write_mode(&mut io)
     }
 
     /// 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, live: bool) -> Result<(), VolumeError> {
+    fn refresh_remote_write_mode(&self, io: &mut VolumeIo) -> Result<(), VolumeError> {
         let has_remote = self.has_remote_file();
         if has_remote {
             // remote-tiered: no writes, deletes allowed
-            if !live {
-                self.no_write.store(NO_WRITE_CAN_DELETE, Ordering::Release);
-            } else {
-                self.mark_tiered_deletable();
-            }
+            self.no_write.store(NO_WRITE_CAN_DELETE, Ordering::Release);
         } 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);
@@ -4414,7 +4363,7 @@ impl Volume {
                     }
                 }
                 *self.volume_info.write().unwrap() = pb_info;
-                self.refresh_remote_write_mode(io, false)?;
+                self.refresh_remote_write_mode(io)?;
                 {
                     let mut vi = self.volume_info.write().unwrap();
                     if vi.version == 0 {
@@ -4445,7 +4394,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, false)?;
+                self.refresh_remote_write_mode(io)?;
                 {
                     let mut vi = self.volume_info.write().unwrap();
                     if vi.version == 0 {
@@ -9752,72 +9701,6 @@ 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

@@ -1,139 +0,0 @@
-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,10 +28,6 @@ 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,16 +81,3 @@ 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, prefix)
+	validateOneEnabledQueue(config)
 
 	for _, queue := range MessageQueues {
-		if queueEnabled(config, prefix, queue) {
+		if config.GetBool(prefix + queue.GetName() + ".enabled") {
 			if err := queue.Initialize(config, prefix+queue.GetName()+"."); err != nil {
 				glog.Fatalf("Failed to initialize notification for %s: %+v",
 					queue.GetName(), err)
@@ -42,26 +42,10 @@ func LoadConfiguration(config *util.ViperProxy, prefix string) {
 
 }
 
-// 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) {
+func validateOneEnabledQueue(config *util.ViperProxy) {
 	enabledQueue := ""
 	for _, queue := range MessageQueues {
-		if queueEnabled(config, prefix, queue) {
+		if config.GetBool(queue.GetName() + ".enabled") {
 			if enabledQueue == "" {
 				enabledQueue = queue.GetName()
 			} else {

weed/notification/webhook/types.go

@@ -3,6 +3,7 @@ package webhook
 import (
 	"fmt"
 	"net/url"
+	"strconv"
 
 	"github.com/seaweedfs/seaweedfs/weed/notification"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -29,6 +30,12 @@ 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,9 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"sort"
-	"strings"
-	"sync"
 	"time"
 
 	"github.com/ThreeDotsLabs/watermill"
@@ -25,145 +22,45 @@ 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 {
-	routers      []*message.Router
+	router       *message.Router
 	queueChannel *gochannel.GoChannel
-	targets      []*target
+	config       *config
+	client       client
+	filter       *filter
 
 	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
 	}
 
-	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)
-			}
-		}
+	if w.filter != nil && !w.filter.shouldPublish(key, eventNotification) {
+		return nil
 	}
-	return firstErr
+
+	m := newWebhookMessage(key, msg)
+	if m == nil {
+		return nil
+	}
+
+	wMsg, err := m.toWaterMillMessage()
+	if err != nil {
+		return err
+	}
+
+	return w.queueChannel.Publish(pubSubTopicName, wMsg)
 }
 
 func (w *webhookMessage) toWaterMillMessage() (*message.Message, error) {
@@ -181,128 +78,105 @@ func (w *webhookMessage) toWaterMillMessage() (*message.Message, error) {
 }
 
 func (w *Queue) Initialize(configuration util.Configuration, prefix string) error {
-	targets, err := buildTargets(configuration, prefix)
-	if err != nil {
+	c := newConfigWithDefaults(configuration, prefix)
+
+	if err := c.validate(); err != nil {
 		return err
 	}
-	if len(targets) == 0 {
-		return fmt.Errorf("webhook endpoint is required")
-	}
 
-	return w.initialize(targets)
+	return w.initialize(c)
 }
 
-func (w *Queue) initialize(targets []*target) error {
+func (w *Queue) initialize(cfg *config) error {
 	w.ctx, w.cancel = context.WithCancel(context.Background())
-	w.targets = targets
+	w.config = cfg
+	w.filter = newFilter(cfg)
 
-	if err := w.setupWatermillQueue(targets); err != nil {
+	// 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 {
 		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(targets []*target) error {
+func (w *Queue) setupWatermillQueue(cfg *config) 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(bufferSize),
+		OutputChannelBuffer: int64(cfg.bufferSize),
 		Persistent:          false,
 	}
 	w.queueChannel = gochannel.NewGoChannel(pubSubConfig, logger)
 
-	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)
+	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
 
-		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
+	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
 
-		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)
-
-		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)
+	poisonQueue, err := middleware.PoisonQueue(w.queueChannel, deadLetterTopic)
+	if err != nil {
+		return fmt.Errorf("failed to create poison queue: %w", err)
 	}
 
-	// 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
+	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() {
-		wg.Wait()
-		w.cancel()
+		// 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)
+		}
+
+		glog.Info("webhook pubsub worker stopped")
 	}()
 
 	return nil
 }
 
-// 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 {
+func (w *Queue) handleWebhook(msg *message.Message) error {
 	// Acquire semaphore slot (blocks if at capacity)
-	t.sem <- struct{}{}
-	defer func() { <-t.sem }()
+	w.sem <- struct{}{}
+	defer func() { <-w.sem }()
 
 	var n filer_pb.EventNotification
 	if err := proto.Unmarshal(msg.Payload, &n); err != nil {
@@ -317,8 +191,8 @@ func (t *target) handleWebhook(msg *message.Message) error {
 		Notification: &n,
 	}
 
-	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)
+	if err := w.client.sendMessage(webhookMsg); err != nil {
+		glog.Errorf("failed to send message to webhook %s: %v", webhookMsg.Key, err)
 		return err
 	}
 
@@ -344,12 +218,10 @@ 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 {
@@ -360,7 +232,7 @@ func (w *Queue) logDeadLetterMessages() error {
 						payload = n.String()
 					}
 				}
-				glog.Errorf("received dead letter message: %s, key: %s, target: %s", payload, key, targetName)
+				glog.Errorf("received dead letter message: %s, key: %s", payload, key)
 			case <-w.ctx.Done():
 				return
 			}

weed/notification/webhook/webhook_queue_test.go

@@ -4,7 +4,6 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"strings"
-	"sync/atomic"
 	"testing"
 	"time"
 
@@ -171,29 +170,6 @@ 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",
@@ -206,19 +182,32 @@ func TestQueueInitialize(t *testing.T) {
 		bufferSize:        100,
 	}
 
-	q := newQueueForTest(t, cfg)
-	defer closeQueueForTest(q)
+	q := &Queue{}
+	err := q.initialize(cfg)
+	if err != nil {
+		t.Errorf("Initialize() error = %v", err)
+	}
 
-	if len(q.routers) != 1 {
+	defer func() {
+		if q.cancel != nil {
+			q.cancel()
+		}
+		time.Sleep(100 * time.Millisecond)
+		if q.router != nil {
+			q.router.Close()
+		}
+	}()
+
+	if q.router == nil {
 		t.Error("Expected router to be initialized")
 	}
 	if q.queueChannel == nil {
 		t.Error("Expected queueChannel to be initialized")
 	}
-	if q.targets[0].client == nil {
+	if q.client == nil {
 		t.Error("Expected client to be initialized")
 	}
-	if q.targets[0].config == nil {
+	if q.config == nil {
 		t.Error("Expected config to be initialized")
 	}
 }
@@ -236,8 +225,21 @@ func TestQueueSendMessage(t *testing.T) {
 		bufferSize:        10,
 	}
 
-	q := newQueueForTest(t, cfg)
-	defer closeQueueForTest(q)
+	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()
+		}
+	}()
 
 	msg := &filer_pb.EventNotification{
 		NewEntry: &filer_pb.Entry{
@@ -245,7 +247,8 @@ func TestQueueSendMessage(t *testing.T) {
 		},
 	}
 
-	if err := q.SendMessage("/test/path", msg); err != nil {
+	err = q.SendMessage("/test/path", msg)
+	if err != nil {
 		t.Errorf("SendMessage() error = %v", err)
 	}
 }
@@ -268,7 +271,7 @@ func TestQueueHandleWebhook(t *testing.T) {
 	}
 
 	client, _ := newHTTPClient(cfg)
-	tgt := &target{
+	q := &Queue{
 		client: client,
 		sem:    make(chan struct{}, cfg.nWorkers),
 	}
@@ -284,7 +287,8 @@ func TestQueueHandleWebhook(t *testing.T) {
 		t.Fatalf("Failed to create watermill message: %v", err)
 	}
 
-	if err := tgt.handleWebhook(wmMsg); err != nil {
+	err = q.handleWebhook(wmMsg)
+	if err != nil {
 		t.Errorf("handleWebhook() error = %v", err)
 	}
 }
@@ -303,8 +307,21 @@ func TestQueueEndToEnd(t *testing.T) {
 		bufferSize:        10,
 	}
 
-	q := newQueueForTest(t, cfg)
-	defer closeQueueForTest(q)
+	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()
+		}
+	}()
 
 	msg := &filer_pb.EventNotification{
 		NewEntry: &filer_pb.Entry{
@@ -312,7 +329,8 @@ func TestQueueEndToEnd(t *testing.T) {
 		},
 	}
 
-	if err := q.SendMessage("/test/path", msg); err != nil {
+	err = q.SendMessage("/test/path", msg)
+	if err != nil {
 		t.Errorf("SendMessage() error = %v", err)
 	}
 }
@@ -329,20 +347,33 @@ func TestQueueRetryMechanism(t *testing.T) {
 		bufferSize:        10,
 	}
 
-	q := newQueueForTest(t, cfg)
-	defer closeQueueForTest(q)
+	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()
+		}
+	}()
 
 	// Verify that the queue is properly configured for retries
-	if q.targets[0].config.maxRetries != 3 {
-		t.Errorf("Expected maxRetries=3, got %d", q.targets[0].config.maxRetries)
+	if q.config.maxRetries != 3 {
+		t.Errorf("Expected maxRetries=3, got %d", q.config.maxRetries)
 	}
 
-	if q.targets[0].config.backoffSeconds != 2 {
-		t.Errorf("Expected backoffSeconds=2, got %d", q.targets[0].config.backoffSeconds)
+	if q.config.backoffSeconds != 2 {
+		t.Errorf("Expected backoffSeconds=2, got %d", q.config.backoffSeconds)
 	}
 
-	if q.targets[0].config.maxBackoffSeconds != 10 {
-		t.Errorf("Expected maxBackoffSeconds=10, got %d", q.targets[0].config.maxBackoffSeconds)
+	if q.config.maxBackoffSeconds != 10 {
+		t.Errorf("Expected maxBackoffSeconds=10, got %d", q.config.maxBackoffSeconds)
 	}
 
 	// Test that we can send a message (retry behavior is handled by Watermill middleware)
@@ -350,7 +381,8 @@ func TestQueueRetryMechanism(t *testing.T) {
 		NewEntry: &filer_pb.Entry{Name: "test.txt"},
 	}
 
-	if err := q.SendMessage("/test/retry", msg); err != nil {
+	err = q.SendMessage("/test/retry", msg)
+	if err != nil {
 		t.Errorf("SendMessage() error = %v", err)
 	}
 }
@@ -387,8 +419,21 @@ func TestQueueNoDuplicateWebhooks(t *testing.T) {
 				bufferSize:        10,
 			}
 
-			q := newQueueForTest(t, cfg)
-			defer closeQueueForTest(q)
+			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()
+				}
+			}()
 
 			// Wait for router and subscriber to be fully ready
 			time.Sleep(200 * time.Millisecond)
@@ -399,7 +444,8 @@ func TestQueueNoDuplicateWebhooks(t *testing.T) {
 				},
 			}
 
-			if err := q.SendMessage("/test/path", msg); err != nil {
+			err = q.SendMessage("/test/path", msg)
+			if err != nil {
 				t.Errorf("SendMessage() error = %v", err)
 			}
 
@@ -561,146 +607,3 @@ 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,7 +25,6 @@ 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"
@@ -1681,7 +1680,6 @@ 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)
 		}
@@ -1721,7 +1719,6 @@ 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)
 		}
@@ -1730,39 +1727,6 @@ 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

@@ -1,250 +0,0 @@
-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,6 +29,7 @@ import (
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
 )
 
@@ -94,6 +95,11 @@ 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,6 +858,11 @@ 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,12 +2,8 @@ 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"
@@ -87,117 +83,6 @@ 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,15 +235,13 @@ func buildPreflightResponse(rule *CORSRule, corsReq *CORSRequest) *CORSResponse
 
 		// Set exposed headers
 		if len(rule.ExposeHeaders) > 0 {
-			response.ExposeHeaders = formatExposeHeaders(rule.ExposeHeaders)
+			response.ExposeHeaders = strings.Join(rule.ExposeHeaders, ", ")
 		}
 
 		// Set max age
 		if rule.MaxAgeSeconds != nil {
 			response.MaxAge = strconv.Itoa(*rule.MaxAgeSeconds)
 		}
-
-		response.AllowCredentials = true
 	}
 
 	return response
@@ -252,8 +250,7 @@ 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,
-		AllowCredentials: true,
+		AllowOrigin: corsReq.Origin,
 	}
 
 	// Set allowed methods
@@ -266,7 +263,7 @@ func buildResponse(rule *CORSRule, corsReq *CORSRequest) *CORSResponse {
 
 	// Set expose headers
 	if len(rule.ExposeHeaders) > 0 {
-		response.ExposeHeaders = formatExposeHeaders(rule.ExposeHeaders)
+		response.ExposeHeaders = strings.Join(rule.ExposeHeaders, ", ")
 	}
 
 	// Set max age
@@ -277,47 +274,6 @@ 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,12 +405,11 @@ func TestEvaluateRequest(t *testing.T) {
 				Method: "GET",
 			},
 			want: &CORSResponse{
-				AllowOrigin:      "http://example.com",
-				AllowMethods:     "GET, POST",
-				AllowHeaders:     "Content-Type",
-				ExposeHeaders:    "ETag",
-				MaxAge:           "3600",
-				AllowCredentials: true,
+				AllowOrigin:   "http://example.com",
+				AllowMethods:  "GET, POST",
+				AllowHeaders:  "Content-Type",
+				ExposeHeaders: "ETag",
+				MaxAge:        "3600",
 			},
 			wantErr: false,
 		},
@@ -422,9 +421,8 @@ func TestEvaluateRequest(t *testing.T) {
 				Method: "PUT",
 			},
 			want: &CORSResponse{
-				AllowOrigin:      "http://other.com",
-				AllowMethods:     "PUT",
-				AllowCredentials: true,
+				AllowOrigin:  "http://other.com",
+				AllowMethods: "PUT",
 			},
 			wantErr: false,
 		},
@@ -449,12 +447,11 @@ 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",
-				AllowCredentials: true,
+				AllowOrigin:   "http://example.com",
+				AllowMethods:  "GET, POST",
+				AllowHeaders:  "Content-Type",
+				ExposeHeaders: "ETag",
+				MaxAge:        "3600",
 			},
 			wantErr: false,
 		},
@@ -470,30 +467,7 @@ func TestEvaluateRequest(t *testing.T) {
 			},
 			want: &CORSResponse{
 				AllowOrigin: "http://example.com",
-				// 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,
+				// No AllowMethods or AllowHeaders because the requested header is forbidden
 			},
 			wantErr: false,
 		},
@@ -537,20 +511,18 @@ func TestApplyHeaders(t *testing.T) {
 		{
 			name: "complete response",
 			corsResp: &CORSResponse{
-				AllowOrigin:      "http://example.com",
-				AllowMethods:     "GET, POST",
-				AllowHeaders:     "Content-Type",
-				ExposeHeaders:    "ETag",
-				MaxAge:           "3600",
-				AllowCredentials: true,
+				AllowOrigin:   "http://example.com",
+				AllowMethods:  "GET, POST",
+				AllowHeaders:  "Content-Type",
+				ExposeHeaders: "ETag",
+				MaxAge:        "3600",
 			},
 			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-Credentials": "true",
+				"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",
 			},
 		},
 		{

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)
 		}
-		logManagerError("CommitTable UpdateTable", err)
+		glog.Errorf("Iceberg: CommitTable UpdateTable error: %v", 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
 		}
-		logManagerError("CreateNamespace", err)
+		glog.Errorf("Iceberg: CreateNamespace error: %v", 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
 			}
-			logManagerError("CommitTransaction UpdateTable", err)
+			glog.Errorf("Iceberg: CommitTransaction UpdateTable error: %v", err)
 			writeManagerError(w, err)
 			return
 		}

weed/s3api/iceberg/log_manager_error_test.go

@@ -1,36 +0,0 @@
-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,9 +158,7 @@ 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).
-// 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) {
+func writeManagerError(w http.ResponseWriter, err error) {
 	switch {
 	case nameValidationError(err),
 		errors.Is(err, iceberg.ErrInvalidSchema),
@@ -168,7 +166,8 @@ func classifyManagerError(err error) (int, string, string) {
 		errors.Is(err, iceberg.ErrInvalidTypeString),
 		errors.Is(err, iceberg.ErrInvalidTransform),
 		errors.Is(err, iceberg.ErrInvalidArgument):
-		return http.StatusBadRequest, "BadRequestException", err.Error()
+		writeError(w, http.StatusBadRequest, "BadRequestException", err.Error())
+		return
 	}
 	// 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
@@ -180,49 +179,31 @@ func classifyManagerError(err error) (int, string, string) {
 			// 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.
-			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)
+			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))
 		case s3tables.ErrCodeNoSuchNamespace:
-			return http.StatusNotFound, "NoSuchNamespaceException", tableErr.Message
+			writeError(w, http.StatusNotFound, "NoSuchNamespaceException", tableErr.Message)
 		case s3tables.ErrCodeNoSuchTable:
-			return http.StatusNotFound, "NoSuchTableException", tableErr.Message
+			writeError(w, http.StatusNotFound, "NoSuchTableException", tableErr.Message)
 		case s3tables.ErrCodeNoSuchView:
-			return http.StatusNotFound, "NoSuchViewException", tableErr.Message
+			writeError(w, http.StatusNotFound, "NoSuchViewException", tableErr.Message)
 		case s3tables.ErrCodeNamespaceAlreadyExists, s3tables.ErrCodeBucketAlreadyExists,
 			s3tables.ErrCodeTableAlreadyExists, s3tables.ErrCodeViewAlreadyExists:
-			return http.StatusConflict, "AlreadyExistsException", tableErr.Message
+			writeError(w, http.StatusConflict, "AlreadyExistsException", tableErr.Message)
 		case s3tables.ErrCodeNamespaceNotEmpty, s3tables.ErrCodeBucketNotEmpty:
-			return http.StatusConflict, "AlreadyExistsException", tableErr.Message
+			writeError(w, http.StatusConflict, "AlreadyExistsException", tableErr.Message)
 		case s3tables.ErrCodeConflict:
-			return http.StatusConflict, "CommitFailedException", tableErr.Message
+			writeError(w, http.StatusConflict, "CommitFailedException", tableErr.Message)
 		case s3tables.ErrCodeAccessDenied:
-			return http.StatusForbidden, "ForbiddenException", tableErr.Message
+			writeError(w, http.StatusForbidden, "ForbiddenException", tableErr.Message)
 		case s3tables.ErrCodeInvalidRequest, s3tables.ErrCodeInvalidIcebergLayout:
-			return http.StatusBadRequest, "BadRequestException", tableErr.Message
+			writeError(w, http.StatusBadRequest, "BadRequestException", tableErr.Message)
+		default:
+			writeError(w, http.StatusInternalServerError, "InternalServerError", err.Error())
 		}
-	}
-	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
 	}
-	logManagerFault("Iceberg: %s error: %v", op, err)
+	writeError(w, http.StatusInternalServerError, "InternalServerError", err.Error())
 }
 
 // resolveWarehouseBucket maps a client-supplied warehouse value to a table

weed/s3api/s3api_object_anonymous_read.go

@@ -171,12 +171,10 @@ 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. The denial is
-// refined like Auth's, so a malformed bucket name answers InvalidBucketName.
+// reaches the wire when the bucket is anonymously listable.
 func (s3a *S3ApiServer) writeAnonymousAwareNotFound(w http.ResponseWriter, r *http.Request, bucket string, code s3err.ErrorCode) {
 	if isAnonymousObjectRead(r) && s3a.authorizeAnonymousBucketList(r, bucket) != s3err.ErrNone {
-		identity, _ := s3_constants.GetIdentityFromContext(r).(*Identity)
-		s3err.WriteErrorResponse(w, r, refineDenialForBucketName(r, identity, s3err.ErrAccessDenied))
+		s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
 		return
 	}
 	s3err.WriteErrorResponse(w, r, code)

weed/s3api/s3api_object_anonymous_read_test.go

@@ -299,9 +299,7 @@ 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},
-		// 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},
+		{"identity deny beats object ACL", true, nil, `{"Effect":"Deny","Action":"s3:GetObject","Resource":"arn:aws:s3:::b/*"}`, false, 403},
 		{"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,10 +181,8 @@ 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, 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() {
+	// Empty directory: 404. A file promoted to a directory keeps its data and stays retrievable.
+	if entry.IsDirectory && totalSize == 0 {
 		s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
 		return 0, 0, false, newStreamErrorWithResponse(fmt.Errorf("directory object %s/%s cannot be retrieved", bucket, object))
 	}
@@ -535,24 +533,6 @@ 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
@@ -565,12 +545,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(statusCode)
+		w.WriteHeader(http.StatusOK)
 		return
 	}
 
 	// Write content
-	w.WriteHeader(statusCode)
+	w.WriteHeader(http.StatusOK)
 	if len(content) > 0 {
 		if _, err := w.Write(content); err != nil {
 			glog.Errorf("serveDirectoryContent: failed to write response: %v", err)
@@ -804,7 +784,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.validateTableBucketObjectPathOrVerify(bucket, object); err != nil {
+	if err := s3a.validateTableBucketObjectPath(bucket, object); err != nil {
 		s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
 		return
 	}
@@ -2499,7 +2479,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.validateTableBucketObjectPathOrVerify(bucket, object); err != nil {
+	if err := s3a.validateTableBucketObjectPath(bucket, object); err != nil {
 		s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
 		return
 	}
@@ -2765,17 +2745,6 @@ 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,21 +2,17 @@ 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"
@@ -1150,9 +1146,7 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
 		return
 	}
 
-	// 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 {
+	if uploadEntryHasSSE(uploadEntry) || sourceEntryIsEncrypted(entry) || uploadEntryHasChecksum(uploadEntry) {
 		etag, sseMetadata, errCode := s3a.copyObjectPartViaReencryption(r, entry, startOffset, endOffset, dstBucket, dstObject, uploadID, partID, uploadEntry)
 		if errCode != s3err.ErrNone {
 			s3err.WriteErrorResponse(w, r, errCode)
@@ -1210,24 +1204,15 @@ 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, partMd5, err := s3a.copyChunksForRange(entry, startOffset, endOffset, dstAssignPath)
+		dstChunks, 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
@@ -1494,7 +1479,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, nil); err != nil {
+		if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, 0, int64(chunk.Size), true /*isFullChunk*/, assignResult); err != nil {
 			return nil, fmt.Errorf("stream chunk: %w", err)
 		}
 		return dstChunk, nil
@@ -1549,22 +1534,9 @@ 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
-		// 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 {
+		if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, offsetInChunk, int64(rangeChunk.Size), isFullChunk, assignResult); 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
 	}
 
@@ -1637,16 +1609,8 @@ func parseRangeHeader(rangeHeader string, fileSize int64) (startOffset, endOffse
 	return startOffset, endOffset, nil
 }
 
-// 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) {
+// 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) {
 	var relevantChunks []*filer_pb.FileChunk
 	var originalChunks []*filer_pb.FileChunk
 
@@ -1677,36 +1641,6 @@ 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
@@ -1730,79 +1664,11 @@ 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, nil, err
+			return nil, err
 		}
 	}
 
-	// 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
+	return dstChunks, 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, nil); err != nil {
+					true /*isFullChunk*/, env.assign); err != nil {
 					b.Fatalf("stream: %v", err)
 				}
 			}

weed/s3api/s3api_object_handlers_copy_etag_test.go

@@ -1,21 +1,14 @@
 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) {
@@ -152,349 +145,3 @@ 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,7 +2,6 @@ package s3api
 
 import (
 	"context"
-	"crypto/md5"
 	"encoding/base64"
 	"encoding/json"
 	"errors"
@@ -420,19 +419,14 @@ 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,19 +73,12 @@ 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.
@@ -173,11 +166,7 @@ func (s3a *S3ApiServer) streamCopyChunkRange(
 			return
 		}
 
-		src := io.Reader(srcResp.Body)
-		if digestWriter != nil && !bodyIsGzipped {
-			src = io.TeeReader(srcResp.Body, digestWriter)
-		}
-		if _, err := io.Copy(fw, src); err != nil {
+		if _, err := io.Copy(fw, srcResp.Body); 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.validateTableBucketObjectPathOrVerify(bucket, object); err != nil {
+	if err := s3a.validateTableBucketObjectPath(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.validateTableBucketObjectPathOrVerify(bucket, object.Key); err != nil {
+			if err := s3a.validateTableBucketObjectPath(bucket, object.Key); err != nil {
 				deleteErrors = append(deleteErrors, deleteErrorFromCode(s3err.ErrAccessDenied, object.Key, object.VersionId))
 				continue
 			}

weed/s3api/s3api_object_handlers_list.go

@@ -540,6 +540,18 @@ 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 != "" {
@@ -794,6 +806,9 @@ 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.
@@ -1009,8 +1024,10 @@ 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, so the
-			// recursion below answers whether it exists and what it holds.
+			// 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
 			if cursor.prefixEndsOnDelimiter {
 				cursor.prefixEndsOnDelimiter = false
 			}
@@ -1019,7 +1036,13 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.
 				// appear as regular keys in recursive listing mode.
 				eachEntryFn(dir, emitEntry)
 			}
-			subNextMarker, subErr := s3a.doListFilerEntries(ctx, client, listDirectoryRequest{dir: dir + "/" + e.Name, marker: subMarker, delimiter: delimiter, bucket: bucket}, cursor, eachEntryFn)
+			// 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)
+			})
 			if subErr != nil {
 				return fmt.Errorf("doListFilerEntries2: %w", subErr)
 			}
@@ -1029,10 +1052,26 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.
 				// keep a resume position that re-enters it.
 				cursor.isTruncated = true
 			}
-			// 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.
+			// 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)
+			}
 			// println("doListFilerEntries2 dir", dir+"/"+e.Name, "subNextMarker", subNextMarker)
 			if cursor.isTruncated || matchedPrefixDir {
 				stopped = true

weed/s3api/s3api_object_handlers_list_directory_test.go

@@ -65,11 +65,12 @@ func TestDirectoryListedAsCommonPrefix(t *testing.T) {
 	assert.Empty(t, seenFiles, "No files should be seen, only the directory")
 }
 
-// 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) {
+// 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) {
 	s3a := &S3ApiServer{
 		option: &S3ApiServerOption{BucketsPath: "/buckets"},
 	}
@@ -86,15 +87,20 @@ func TestEmptyDirectoryNotListedUnderOwnPrefix(t *testing.T) {
 		},
 	}
 
-	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)
+	// 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")
 	}
 }
 
@@ -179,10 +185,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 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) {
+// 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) {
 	filer := &fakeBucketDeleteFiler{entriesByDir: map[string][]*filer_pb.Entry{
 		"/buckets/test": {
 			{Name: "locks", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mime: ""}},
@@ -202,5 +208,6 @@ func TestEmptyDirectoryHiddenInDelimitedSelfListing(t *testing.T) {
 	})
 	require.NoError(t, err)
 	assert.Empty(t, response.Contents, "an empty directory is not a readable key")
-	assert.Empty(t, response.CommonPrefixes, "an empty directory has no prefixes to fold into")
+	require.Len(t, response.CommonPrefixes, 1)
+	assert.Equal(t, "locks/", response.CommonPrefixes[0].Prefix)
 }

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, refineDenialForBucketName(r, identity, errCode))
+		s3err.WriteErrorResponse(w, r, 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, refineDenialForBucketName(r, identity, errCode))
+		s3err.WriteErrorResponse(w, r, errCode)
 		return
 	}
 
@@ -362,24 +362,10 @@ 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 {
-		identity, errCode = iam.doesPolicySignatureV2Match(formValues)
-	} else {
-		identity, errCode = iam.doesPolicySignatureV4Match(formValues)
+		return iam.doesPolicySignatureV2Match(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
+	return iam.doesPolicySignatureV4Match(formValues)
 }

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 isVerify, err := s3a.validateTableBucketObjectPathOrVerify(bucket, object); err != nil || (isVerify && !volumeVerifyPutAllowed(r, object)) {
+	if err := s3a.validateTableBucketObjectPath(bucket, object); err != nil {
 		s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
 		return
 	}
@@ -991,11 +991,7 @@ 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)
-			// 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)) {
+			if errors.Is(err, filer_pb.ErrExistingIsDirectory) && !isReservedDirectoryName(entry.Name) {
 				// 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,33 +40,3 @@ 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,10 +661,7 @@ 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) {
-		// 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))
+		return s3a.authorizeAnonymousObjectRead(r, bucket, object, objectEntry)
 	}
 	identityRaw := GetIdentityFromContext(r)
 	var identity *Identity

weed/s3api/s3tables/filer_ops.go

@@ -262,6 +262,14 @@ func (h *S3TablesHandler) getExtendedAttribute(ctx context.Context, client filer
 	return data, nil
 }
 
+// absentEntry reports whether err means there is no usable catalog entry at
+// the path: nothing is there, or what is there carries none of the attributes
+// the operation keys on (a plain S3 bucket holding the name, a directory left
+// without metadata).
+func absentEntry(err error) bool {
+	return errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, ErrAttributeNotFound)
+}
+
 // loadNamespaceMetadata resolves a namespace to its metadata. A directory that
 // carries no namespace metadata is not a namespace, so a missing attribute
 // reports the same absence as a missing entry and every caller tests one

weed/s3api/s3tables/handler.go

@@ -127,10 +127,6 @@ 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
@@ -227,51 +223,10 @@ func (h *S3TablesHandler) HandleRequest(w http.ResponseWriter, r *http.Request,
 	}
 
 	if err != nil {
-		logRequestError(operation, sw.status, err)
+		glog.Errorf("S3Tables: error handling %s: %v", operation, 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.
@@ -504,68 +459,18 @@ 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) {
+// mayLearnAbsence reports whether the caller may be shown that an operation's
+// target is absent rather than the operation's denial: the admin, a trusted
+// handler, or a grant the request itself carries (an identity-level action
+// grant, or the open default-allow context).
+func (h *S3TablesHandler) mayLearnAbsence(r *http.Request, operation, bucketName string) bool {
+	if h.getAccountID(r) == s3_constants.AccountAdminId || h.defaultAllowFor(r) {
 		return true
 	}
-	return bucketOwner == "" && checkPermission(operation, principal, "", "", resourceARN, ctx)
+	return hasIdentityPermission(operation, &PolicyContext{
+		TableBucketName: bucketName,
+		IdentityActions: getIdentityActions(r),
+	})
 }
 
 func (h *S3TablesHandler) readTags(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) (map[string]string, error) {

weed/s3api/s3tables/handler_absent_target_test.go

@@ -0,0 +1,246 @@
+package s3tables
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func runAbsentOp(t *testing.T, m *Manager, fs *s3tablestest.MemFiler, operation string, input interface{}, name string, actions []string) s3TablesHTTPError {
+	t.Helper()
+	body, err := json.Marshal(input)
+	require.NoError(t, err)
+	identity := &testIdentity{Name: name, Account: &testIdentityAccount{Id: name}, Actions: actions}
+	req, err := newManagerRequest(s3_constants.SetIdentityInContext(context.Background(), identity), operation, body, name)
+	require.NoError(t, err)
+
+	recorder := httptest.NewRecorder()
+	m.handler.HandleRequest(recorder, req, NewManagerClient(fs.Client))
+	result := recorder.Result()
+	defer result.Body.Close()
+
+	var response S3TablesError
+	if result.StatusCode >= 400 {
+		require.NoError(t, json.NewDecoder(result.Body).Decode(&response))
+	}
+	return s3TablesHTTPError{status: result.StatusCode, body: response}
+}
+
+func absentBucketARN(t *testing.T, name string) string {
+	t.Helper()
+	arn, err := BuildBucketARN(DefaultRegion, DefaultAccountID, name)
+	require.NoError(t, err)
+	return arn
+}
+
+func absentTableARN(t *testing.T, bucket, namespace, table string) string {
+	t.Helper()
+	arn, err := BuildTableARN(DefaultRegion, DefaultAccountID, bucket, namespace, table)
+	require.NoError(t, err)
+	return arn
+}
+
+// An outsider gets the operation's own denial for an absent name, a name a
+// plain S3 bucket holds, and a name it is denied on, so none of them can be
+// told apart.
+func TestAbsentTargetsAnswerLikeDenials(t *testing.T) {
+	fs, m := startRenameManager(t)
+	m.SetTrusted(false)
+	m.SetDefaultAllow(false)
+
+	ghostARN := absentBucketARN(t, "ghostbkt")
+	plainARN := absentBucketARN(t, "plainbkt")
+	fs.Put(TablesPath, "plainbkt", nil) // a plain S3 bucket holds the name
+
+	outsider := []string{"Read"}
+
+	t.Run("ListTables", func(t *testing.T) {
+		for _, tc := range []struct {
+			name string
+			req  *ListTablesRequest
+		}{
+			{"denied namespace", &ListTablesRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}}},
+			{"missing namespace", &ListTablesRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ghost"}}},
+			{"namespace in missing bucket", &ListTablesRequest{TableBucketARN: ghostARN, Namespace: []string{"ns"}}},
+			{"namespace in plain bucket", &ListTablesRequest{TableBucketARN: plainARN, Namespace: []string{"ns"}}},
+			{"denied bucket", &ListTablesRequest{TableBucketARN: mustBucketARN(t)}},
+			{"missing bucket", &ListTablesRequest{TableBucketARN: ghostARN}},
+			{"plain bucket", &ListTablesRequest{TableBucketARN: plainARN}},
+		} {
+			got := runAbsentOp(t, m, fs, "ListTables", tc.req, "attacker", outsider)
+			assert.Equal(t, 403, got.status, "%s: %s", tc.name, got.body.Message)
+			assert.Equal(t, ErrCodeAccessDenied, got.body.Type, tc.name)
+		}
+	})
+
+	t.Run("ListViews", func(t *testing.T) {
+		for _, tc := range []struct {
+			name string
+			req  *ListViewsRequest
+		}{
+			{"denied namespace", &ListViewsRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}}},
+			{"missing namespace", &ListViewsRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ghost"}}},
+			{"missing bucket", &ListViewsRequest{TableBucketARN: ghostARN, Namespace: []string{"ns"}}},
+			{"plain bucket", &ListViewsRequest{TableBucketARN: plainARN, Namespace: []string{"ns"}}},
+		} {
+			got := runAbsentOp(t, m, fs, "ListViews", tc.req, "attacker", outsider)
+			assert.Equal(t, 403, got.status, "%s: %s", tc.name, got.body.Message)
+			assert.Equal(t, ErrCodeAccessDenied, got.body.Type, tc.name)
+		}
+	})
+
+	t.Run("CreateNamespace", func(t *testing.T) {
+		for _, tc := range []struct {
+			name string
+			arn  string
+		}{
+			{"denied bucket", mustBucketARN(t)},
+			{"missing bucket", ghostARN},
+			{"plain bucket", plainARN},
+		} {
+			got := runAbsentOp(t, m, fs, "CreateNamespace", &CreateNamespaceRequest{TableBucketARN: tc.arn, Namespace: []string{"ns"}}, "attacker", outsider)
+			assert.Equal(t, 403, got.status, "%s: %s", tc.name, got.body.Message)
+			assert.Equal(t, ErrCodeAccessDenied, got.body.Type, tc.name)
+		}
+	})
+
+	t.Run("GetTableBucket and ListNamespaces", func(t *testing.T) {
+		for _, tc := range []struct {
+			name string
+			arn  string
+		}{
+			{"denied bucket", mustBucketARN(t)},
+			{"missing bucket", ghostARN},
+			{"plain bucket", plainARN},
+		} {
+			for _, op := range []string{"GetTableBucket", "ListNamespaces"} {
+				var req interface{} = &GetTableBucketRequest{TableBucketARN: tc.arn}
+				if op == "ListNamespaces" {
+					req = &ListNamespacesRequest{TableBucketARN: tc.arn}
+				}
+				got := runAbsentOp(t, m, fs, op, req, "attacker", outsider)
+				assert.Equal(t, 404, got.status, "%s %s: %s", op, tc.name, got.body.Message)
+				assert.Equal(t, ErrCodeNoSuchBucket, got.body.Type, "%s %s", op, tc.name)
+			}
+		}
+	})
+
+	t.Run("tag operations", func(t *testing.T) {
+		missingTableARN := absentTableARN(t, renameTestBucket, "ns", "ghost")
+		deniedTableARN := absentTableARN(t, renameTestBucket, "ns", "t")
+		for _, tc := range []struct {
+			name string
+			op   string
+			req  interface{}
+		}{
+			{"TagResource denied table", "TagResource", &TagResourceRequest{ResourceARN: deniedTableARN, Tags: map[string]string{"k": "v"}}},
+			{"TagResource missing table", "TagResource", &TagResourceRequest{ResourceARN: missingTableARN, Tags: map[string]string{"k": "v"}}},
+			{"TagResource missing bucket", "TagResource", &TagResourceRequest{ResourceARN: ghostARN, Tags: map[string]string{"k": "v"}}},
+			{"ListTagsForResource denied table", "ListTagsForResource", &ListTagsForResourceRequest{ResourceARN: deniedTableARN}},
+			{"ListTagsForResource missing table", "ListTagsForResource", &ListTagsForResourceRequest{ResourceARN: missingTableARN}},
+			{"ListTagsForResource missing bucket", "ListTagsForResource", &ListTagsForResourceRequest{ResourceARN: ghostARN}},
+			{"UntagResource denied table", "UntagResource", &UntagResourceRequest{ResourceARN: deniedTableARN, TagKeys: []string{"k"}}},
+			{"UntagResource missing table", "UntagResource", &UntagResourceRequest{ResourceARN: missingTableARN, TagKeys: []string{"k"}}},
+			{"UntagResource missing bucket", "UntagResource", &UntagResourceRequest{ResourceARN: ghostARN, TagKeys: []string{"k"}}},
+		} {
+			got := runAbsentOp(t, m, fs, tc.op, tc.req, "attacker", outsider)
+			assert.Equal(t, 403, got.status, "%s: %s", tc.name, got.body.Message)
+			assert.Equal(t, ErrCodeAccessDenied, got.body.Type, tc.name)
+		}
+	})
+}
+
+// A caller the operation would have authorized — the bucket owner, a bucket-
+// policy grantee, an identity-level grant, the admin or a trusted handler —
+// may still be told the name is absent: the empty list or the not-found stays.
+func TestGrantedCallersLearnAbsence(t *testing.T) {
+	fs, m := startRenameManager(t)
+	m.SetTrusted(false)
+	m.SetDefaultAllow(false)
+
+	ghostARN := absentBucketARN(t, "ghostbkt")
+	missingTableARN := absentTableARN(t, renameTestBucket, "ns", "ghost")
+	adminActions := []string{s3_constants.ACTION_ADMIN}
+
+	// The bucket owner, and a grantee the bucket policy names, learn absence.
+	bucketGrant := fmt.Sprintf(`{"Statement":[{"Effect":"Allow","Principal":"%s","Action":"s3tables:*","Resource":"*"}]}`, "grantee")
+	fs.Get(TablesPath, renameTestBucket).Extended[ExtendedKeyPolicy] = []byte(bucketGrant)
+
+	t.Run("bucket grant learns an absent namespace", func(t *testing.T) {
+		req := &ListTablesRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ghost"}}
+		for _, name := range []string{DefaultAccountID, "grantee"} {
+			got := runAbsentOp(t, m, fs, "ListTables", req, name, nil)
+			assert.Equal(t, 200, got.status, "%s", name)
+		}
+		// And on a namespace that exists they still see the table.
+		got := runAbsentOp(t, m, fs, "ListTables", &ListTablesRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}}, "grantee", nil)
+		assert.Equal(t, 200, got.status)
+	})
+
+	t.Run("identity grant learns an absent bucket", func(t *testing.T) {
+		got := runAbsentOp(t, m, fs, "ListTables", &ListTablesRequest{TableBucketARN: ghostARN}, "lister", []string{"s3tables:ListTables"})
+		assert.Equal(t, 200, got.status)
+		got = runAbsentOp(t, m, fs, "ListViews", &ListViewsRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ghost"}}, "lister", []string{"s3tables:ListViews"})
+		assert.Equal(t, 200, got.status)
+	})
+
+	t.Run("admin learns an absent bucket", func(t *testing.T) {
+		got := runAbsentOp(t, m, fs, "ListTables", &ListTablesRequest{TableBucketARN: ghostARN}, s3_constants.AccountAdminId, adminActions)
+		assert.Equal(t, 200, got.status)
+		got = runAbsentOp(t, m, fs, "CreateNamespace", &CreateNamespaceRequest{TableBucketARN: ghostARN, Namespace: []string{"ns"}}, s3_constants.AccountAdminId, adminActions)
+		assert.Equal(t, 404, got.status)
+		assert.Equal(t, ErrCodeNoSuchBucket, got.body.Type)
+	})
+
+	t.Run("tag operations answer 404 to granted callers", func(t *testing.T) {
+		got := runAbsentOp(t, m, fs, "TagResource", &TagResourceRequest{ResourceARN: missingTableARN, Tags: map[string]string{"k": "v"}}, "grantee", nil)
+		assert.Equal(t, 404, got.status, got.body.Message)
+		assert.Equal(t, ErrCodeNoSuchTable, got.body.Type)
+		got = runAbsentOp(t, m, fs, "ListTagsForResource", &ListTagsForResourceRequest{ResourceARN: missingTableARN}, DefaultAccountID, nil)
+		assert.Equal(t, 404, got.status)
+		got = runAbsentOp(t, m, fs, "UntagResource", &UntagResourceRequest{ResourceARN: missingTableARN, TagKeys: []string{"k"}}, s3_constants.AccountAdminId, adminActions)
+		assert.Equal(t, 404, got.status)
+	})
+
+	t.Run("a trusted handler still lists an absent namespace", func(t *testing.T) {
+		m.SetTrusted(true)
+		got := runAbsentOp(t, m, fs, "ListTables", &ListTablesRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ghost"}}, "svc", nil)
+		assert.Equal(t, 200, got.status)
+	})
+}
+
+// A bucket-policy Deny still vetoes the operation on an absent name: a denied
+// caller must not trade the denial for a not-found or an empty list.
+func TestAbsentResourceBucketDenyStillDenies(t *testing.T) {
+	fs, m := startRenameManager(t)
+	m.SetTrusted(false)
+	m.SetDefaultAllow(false)
+
+	deny := `{"Statement":[{"Effect":"Deny","Principal":"lister","Action":"s3tables:*","Resource":"*"}]}`
+	fs.Get(TablesPath, renameTestBucket).Extended[ExtendedKeyPolicy] = []byte(deny)
+
+	got := runAbsentOp(t, m, fs, "TagResource",
+		&TagResourceRequest{ResourceARN: absentTableARN(t, renameTestBucket, "ns", "ghost"), Tags: map[string]string{"k": "v"}},
+		"lister", []string{"s3tables:TagResource"})
+	assert.Equal(t, 403, got.status, got.body.Message)
+	assert.Equal(t, ErrCodeAccessDenied, got.body.Type)
+
+	for _, tc := range []struct {
+		op  string
+		req interface{}
+	}{
+		{"ListTables", &ListTablesRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ghost"}}},
+		{"ListTables", &ListTablesRequest{TableBucketARN: mustBucketARN(t)}},
+		{"ListViews", &ListViewsRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ghost"}}},
+	} {
+		got := runAbsentOp(t, m, fs, tc.op, tc.req, "lister", []string{"s3tables:ListTables", "s3tables:ListViews"})
+		assert.Equal(t, 403, got.status, "%s: %s", tc.op, got.body.Message)
+	}
+}

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) || errors.Is(err, ErrAttributeNotFound) {
+		if absentEntry(err) {
 			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,7 +3,6 @@ package s3tables
 import (
 	"context"
 	"encoding/json"
-	"fmt"
 	"net/http/httptest"
 	"testing"
 
@@ -212,165 +211,3 @@ 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

@@ -1,123 +0,0 @@
-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,9 +36,10 @@ 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.V(3).Infof("S3Tables: handleCreateNamespace called")
+	glog.Errorf("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
 	}
@@ -139,7 +140,9 @@ 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, "", "") {
+			// An absent name and a denied one must read alike; the caller is
+			// told the bucket is missing only when it may learn that.
+			if !h.mayLearnAbsence(r, "CreateNamespace", bucketName) {
 				h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to create namespace in this bucket")
 				return ErrAccessDenied
 			}
@@ -484,7 +487,7 @@ func (h *S3TablesHandler) handleListNamespaces(w http.ResponseWriter, r *http.Re
 	})
 
 	if err != nil {
-		if errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, ErrAttributeNotFound) {
+		if absentEntry(err) {
 			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

@@ -1,39 +0,0 @@
-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,10 +659,16 @@ 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)
+			if !absentEntry(err) {
+				return err
 			}
-			return err
+			ctx := &PolicyContext{RequestTags: req.Tags, TagKeys: requestTagKeys}
+			if rType == ResourceTypeTable {
+				if _, ctx.Namespace, ctx.TableName, err = parseTableFromARN(req.ResourceARN); err != nil {
+					return err
+				}
+			}
+			return h.authorizeAbsentResource(r, client, "TagResource", req.ResourceARN, resourcePath, "not authorized to tag resource", ctx)
 		}
 
 		ownerAccountID, bucketName, err := h.extractResourceOwnerAndBucket(data, resourcePath, rType)
@@ -796,10 +802,16 @@ 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)
+			if !absentEntry(err) {
+				return err
 			}
-			return err
+			ctx := &PolicyContext{}
+			if rType == ResourceTypeTable {
+				if _, ctx.Namespace, ctx.TableName, err = parseTableFromARN(req.ResourceARN); err != nil {
+					return err
+				}
+			}
+			return h.authorizeAbsentResource(r, client, "ListTagsForResource", req.ResourceARN, resourcePath, "not authorized to list tags for resource", ctx)
 		}
 
 		ownerAccountID, bucketName, err := h.extractResourceOwnerAndBucket(data, resourcePath, rType)
@@ -919,10 +931,16 @@ 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)
+			if !absentEntry(err) {
+				return err
 			}
-			return err
+			ctx := &PolicyContext{TagKeys: req.TagKeys}
+			if rType == ResourceTypeTable {
+				if _, ctx.Namespace, ctx.TableName, err = parseTableFromARN(req.ResourceARN); err != nil {
+					return err
+				}
+			}
+			return h.authorizeAbsentResource(r, client, "UntagResource", req.ResourceARN, resourcePath, "not authorized to untag resource", ctx)
 		}
 
 		ownerAccountID, bucketName, err := h.extractResourceOwnerAndBucket(data, resourcePath, rType)
@@ -1026,18 +1044,58 @@ 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)
+// authorizeAbsentResource answers a tag operation whose target carries no
+// readable metadata. A caller holding a grant on the bucket that owns the
+// name, or an identity-level or admin grant, is told the resource is absent;
+// anyone else is refused with the operation's denial.
+func (h *S3TablesHandler) authorizeAbsentResource(r *http.Request, client filer_pb.SeaweedFilerClient, operation, resourceARN, resourcePath, message string, ctx *PolicyContext) error {
+	parts := strings.Split(strings.Trim(resourcePath, "/"), "/")
+	if len(parts) < 2 {
+		return filer_pb.ErrNotFound
 	}
-	if !h.authorizeAbsentTarget(r, &singleClient{client}, operation, bucketName, namespace, name, resourceARN) {
-		return NewAuthError(operation, h.getAccountID(r), deniedMessage)
+	bucketName := parts[1]
+	bucketPath := GetTableBucketPath(bucketName)
+	ctx.TableBucketName = bucketName
+
+	var bucketMeta tableBucketMetadata
+	var bucketPolicy string
+	if data, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata); err == nil {
+		if err := json.Unmarshal(data, &bucketMeta); err != nil {
+			return fmt.Errorf("failed to unmarshal bucket metadata: %w", err)
+		}
+	} else if !absentEntry(err) {
+		return err
+	}
+	if policy, err := h.readPolicy(r, client, bucketPath); err == nil {
+		bucketPolicy = policy
+	} else if !absentEntry(err) {
+		return err
+	}
+	if tags, err := h.readTags(r.Context(), client, bucketPath); err == nil {
+		if tags != nil {
+			ctx.TableBucketTags = tags
+		}
+	} else if !absentEntry(err) {
+		return err
+	}
+
+	principal := h.getAccountID(r)
+	ctx.TrustedIssuers = h.trustedIssuers(r)
+	ctx.IsLocalAccountID = h.localAccountIDs
+	ctx.IdentityActions = getIdentityActions(r)
+	ctx.DefaultAllow = h.defaultAllowFor(r)
+	// Evaluate against the canonical ARN, not the request's: the ARN parser
+	// accepts any account, so a crafted ARN could otherwise dodge a Deny's
+	// resource match.
+	if ctx.TableName != "" {
+		resourceARN = h.generateTableARN(bucketMeta.OwnerAccountID, bucketName, ctx.Namespace+"/"+ctx.TableName)
+	} else {
+		resourceARN = h.generateTableBucketARN(bucketMeta.OwnerAccountID, bucketName)
+	}
+	if deniedByPolicy(operation, principal, bucketPolicy, resourceARN, ctx) ||
+		(!CheckPermissionWithContext(operation, principal, bucketMeta.OwnerAccountID, bucketPolicy, resourceARN, ctx) &&
+			!h.mayLearnAbsence(r, operation, bucketName)) {
+		return NewAuthError(operation, principal, message)
 	}
 	return filer_pb.ErrNotFound
 }

weed/s3api/s3tables/handler_table.go

@@ -514,73 +514,69 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 			var namespacePolicy, bucketPolicy string
 			bucketTags := map[string]string{}
 
-			// Fetch namespace metadata and policy
-			targetAbsent := false
+			// Fetch namespace metadata and policy; a missing or foreign entry
+			// leaves them empty and lets the check below answer.
 			data, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyMetadata)
-			if err != nil {
-				if !errors.Is(err, filer_pb.ErrNotFound) && !errors.Is(err, ErrAttributeNotFound) {
+			nsFound := err == nil
+			if nsFound {
+				if err := json.Unmarshal(data, &nsMeta); err != nil {
 					return err
 				}
-				targetAbsent = true
-			} else if err := json.Unmarshal(data, &nsMeta); err != nil {
+				policyData, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyPolicy)
+				if err == nil {
+					namespacePolicy = string(policyData)
+				} else if !errors.Is(err, ErrAttributeNotFound) {
+					return fmt.Errorf("failed to fetch namespace policy: %w", err)
+				}
+			} else if !absentEntry(err) {
 				return err
 			}
 
-			// Fetch namespace policy if it exists
-			policyData, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyPolicy)
-			if err == nil {
-				namespacePolicy = string(policyData)
-			} else if !errors.Is(err, ErrAttributeNotFound) && !errors.Is(err, filer_pb.ErrNotFound) {
-				return fmt.Errorf("failed to fetch namespace policy: %w", err)
-			}
-
 			// Fetch bucket metadata and policy
 			data, err = h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata)
 			if err == nil {
 				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) {
+			} else if !absentEntry(err) {
 				return fmt.Errorf("failed to fetch bucket metadata: %w", err)
 			}
 
-			policyData, err = h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
+			policyData, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
 			if err == nil {
 				bucketPolicy = string(policyData)
-			} else if !errors.Is(err, ErrAttributeNotFound) && !errors.Is(err, filer_pb.ErrNotFound) {
+			} else if !absentEntry(err) {
 				return fmt.Errorf("failed to fetch bucket policy: %w", err)
 			}
 			if tags, err := h.readTags(r.Context(), client, bucketPath); err != nil {
-				if !errors.Is(err, filer_pb.ErrNotFound) {
+				if !absentEntry(err) {
 					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)
+			policyCtx := &PolicyContext{
+				TrustedIssuers:   h.trustedIssuers(r),
+				IsLocalAccountID: h.localAccountIDs,
+				TableBucketName:  bucketName,
+				Namespace:        namespaceName,
+				TableBucketTags:  bucketTags,
+				IdentityActions:  identityActions,
+				DefaultAllow:     h.defaultAllowFor(r),
+			}
 			if !CheckTableOrBucketPermission("ListTables", accountID,
 				nsMeta.OwnerAccountID, namespacePolicy, bucketARN,
-				bucketMeta.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
-					TrustedIssuers:   h.trustedIssuers(r),
-					IsLocalAccountID: h.localAccountIDs,
-					TableBucketName:  bucketName,
-					Namespace:        namespaceName,
-					TableBucketTags:  bucketTags,
-					IdentityActions:  identityActions,
-					DefaultAllow:     h.defaultAllowFor(r),
-				}) {
-				return ErrAccessDenied
+				bucketMeta.OwnerAccountID, bucketPolicy, bucketARN, policyCtx) {
+				// The target is absent: an empty list would reveal that, so
+				// only a caller that may learn absence proceeds to it, and a
+				// resource-policy Deny vetoes even that.
+				if nsFound || !h.mayLearnAbsence(r, "ListTables", bucketName) ||
+					deniedByPolicy("ListTables", accountID, bucketPolicy, bucketARN, policyCtx) {
+					return ErrAccessDenied
+				}
 			}
 
 			tables, paginationToken, err = h.listTablesInNamespaceWithClient(r, client, bucketName, namespaceName, req.Prefix, req.ContinuationToken, maxTables)
@@ -594,45 +590,47 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 			var bucketPolicy string
 			bucketTags := map[string]string{}
 
-			// Fetch bucket metadata and policy
+			// Fetch bucket metadata and policy; a missing or foreign entry
+			// leaves them empty and lets the check below answer.
 			data, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata)
-			if err != nil {
-				if !errors.Is(err, filer_pb.ErrNotFound) && !errors.Is(err, ErrAttributeNotFound) {
-					return fmt.Errorf("failed to fetch bucket metadata: %w", err)
+			bucketFound := err == nil
+			if bucketFound {
+				if err := json.Unmarshal(data, &bucketMeta); err != nil {
+					return fmt.Errorf("failed to unmarshal 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)
+			} else if !absentEntry(err) {
+				return fmt.Errorf("failed to fetch bucket metadata: %w", err)
 			}
 
-			// Fetch bucket policy if it exists
 			policyData, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
 			if err == nil {
 				bucketPolicy = string(policyData)
-			} else if !errors.Is(err, ErrAttributeNotFound) {
+			} else if !absentEntry(err) {
 				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 !absentEntry(err) {
+					return fmt.Errorf("failed to read bucket tags: %w", err)
+				}
 			} else if tags != nil {
 				bucketTags = tags
 			}
 
 			bucketARN := h.generateTableBucketARN(bucketMeta.OwnerAccountID, bucketName)
 			identityActions := getIdentityActions(r)
-			if !CheckPermissionWithContext("ListTables", accountID, bucketMeta.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
+			policyCtx := &PolicyContext{
 				TrustedIssuers:   h.trustedIssuers(r),
 				IsLocalAccountID: h.localAccountIDs,
 				TableBucketName:  bucketName,
 				TableBucketTags:  bucketTags,
 				IdentityActions:  identityActions,
 				DefaultAllow:     h.defaultAllowFor(r),
-			}) {
-				return ErrAccessDenied
+			}
+			if !CheckPermissionWithContext("ListTables", accountID, bucketMeta.OwnerAccountID, bucketPolicy, bucketARN, policyCtx) {
+				if bucketFound || !h.mayLearnAbsence(r, "ListTables", bucketName) ||
+					deniedByPolicy("ListTables", accountID, bucketPolicy, bucketARN, policyCtx) {
+					return ErrAccessDenied
+				}
 			}
 
 			tables, paginationToken, err = h.listTablesInAllNamespaces(r, client, bucketName, req.Prefix, req.ContinuationToken, maxTables)

weed/s3api/s3tables/handler_view.go

@@ -278,16 +278,12 @@ func (h *S3TablesHandler) handleListViews(w http.ResponseWriter, r *http.Request
 		bucketPath := GetTableBucketPath(bucketName)
 		var nsMeta namespaceMetadata
 		data, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyMetadata)
-		if err != nil {
-			if !errors.Is(err, filer_pb.ErrNotFound) && !errors.Is(err, ErrAttributeNotFound) {
+		nsFound := err == nil
+		if nsFound {
+			if err := json.Unmarshal(data, &nsMeta); err != nil {
 				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 {
+		} else if !absentEntry(err) {
 			return err
 		}
 
@@ -296,8 +292,23 @@ func (h *S3TablesHandler) handleListViews(w http.ResponseWriter, r *http.Request
 			return err
 		}
 		bucketARN := h.generateTableBucketARN(bucketMeta.OwnerAccountID, bucketName)
-		if !h.authorizeViewOp(r, "ListViews", h.getAccountID(r), nsMeta.OwnerAccountID, bucketMeta.OwnerAccountID, namespacePolicy, bucketPolicy, bucketARN, bucketName, namespaceName, "", bucketTags) {
-			return ErrAccessDenied
+		accountID := h.getAccountID(r)
+		if !h.authorizeViewOp(r, "ListViews", accountID, nsMeta.OwnerAccountID, bucketMeta.OwnerAccountID, namespacePolicy, bucketPolicy, bucketARN, bucketName, namespaceName, "", bucketTags) {
+			// An empty list would reveal the namespace is absent; only a
+			// caller that may learn absence proceeds to it, and a
+			// resource-policy Deny vetoes even that.
+			if nsFound || !h.mayLearnAbsence(r, "ListViews", bucketName) ||
+				deniedByPolicy("ListViews", accountID, bucketPolicy, bucketARN, &PolicyContext{
+					TrustedIssuers:   h.trustedIssuers(r),
+					IsLocalAccountID: h.localAccountIDs,
+					TableBucketName:  bucketName,
+					Namespace:        namespaceName,
+					TableBucketTags:  bucketTags,
+					IdentityActions:  getIdentityActions(r),
+					DefaultAllow:     h.defaultAllowFor(r),
+				}) {
+				return ErrAccessDenied
+			}
 		}
 
 		views, paginationToken, err = h.listViewsInNamespace(r, client, bucketName, namespaceName, req.Prefix, req.ContinuationToken, maxViews)
@@ -643,25 +654,27 @@ func (h *S3TablesHandler) loadAuthContext(r *http.Request, filerClient FilerClie
 			if e := json.Unmarshal(data, &bucketMetadata); e != nil {
 				return fmt.Errorf("failed to unmarshal bucket metadata: %w", e)
 			}
-		} else if !errors.Is(e, ErrAttributeNotFound) {
+		} else if !absentEntry(e) {
 			return fmt.Errorf("failed to fetch bucket metadata: %w", e)
 		}
 
 		policyData, e := h.getExtendedAttribute(r.Context(), client, resourcePath, ExtendedKeyPolicy)
 		if e == nil {
 			resourcePolicy = string(policyData)
-		} else if !errors.Is(e, ErrAttributeNotFound) {
+		} else if !absentEntry(e) {
 			return fmt.Errorf("failed to fetch resource policy: %w", e)
 		}
 
 		policyData, e = h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
 		if e == nil {
 			bucketPolicy = string(policyData)
-		} else if !errors.Is(e, ErrAttributeNotFound) {
+		} else if !absentEntry(e) {
 			return fmt.Errorf("failed to fetch bucket policy: %w", e)
 		}
 		if tags, e := h.readTags(r.Context(), client, bucketPath); e != nil {
-			return e
+			if !absentEntry(e) {
+				return e
+			}
 		} else if tags != nil {
 			bucketTags = tags
 		}

weed/s3api/table_bucket_verify_object_test.go

@@ -1,163 +0,0 @@
-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,22 +54,6 @@ 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
@@ -90,10 +74,7 @@ func (vs *VolumeServer) getCurrentMaster() pb.ServerAddress {
 }
 
 // setCurrentMaster updates vs.currentMaster under a write lock. The
-// 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 "".
+// heartbeat goroutine calls this whenever it (re)connects to a master.
 func (vs *VolumeServer) setCurrentMaster(master pb.ServerAddress) {
 	vs.currentMasterLock.Lock()
 	vs.currentMaster = master
@@ -141,6 +122,7 @@ 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)
@@ -154,15 +136,12 @@ func (vs *VolumeServer) heartbeat() {
 				} else {
 					// Regular error, reset duplicate retry count
 					duplicateRetryCount = 0
-					// 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))
+					time.Sleep(vs.pulsePeriod)
 				}
 
 				stats.VolumeServerMasterDisconnections.WithLabelValues(master.String()).Inc()
 				newLeader = ""
+				vs.store.MasterAddress = ""
 			} else {
 				// Successful connection, reset retry count
 				duplicateRetryCount = 0
@@ -201,42 +180,17 @@ 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 {
@@ -277,9 +231,9 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp
 			}
 			if volumeOptsChanged {
 				if vs.store.MaybeAdjustVolumeMax() {
-					select {
-					case resendChan <- struct{}{}:
-					default:
+					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
 					}
 				}
 			}
@@ -290,6 +244,15 @@ 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
+				}
+			}
 		}
 	}()
 
@@ -396,11 +359,6 @@ 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,17 +4,11 @@ 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"
 )
@@ -66,279 +60,3 @@ 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,8 +110,6 @@ 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

@@ -1,77 +0,0 @@
-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,7 +148,6 @@ 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,7 +3,6 @@ package s3_backend
 import (
 	"fmt"
 	"io"
-	"math"
 	"os"
 	"strconv"
 	"strings"
@@ -45,7 +44,6 @@ type S3BackendStorage struct {
 	forcePathStyle        bool
 	uploadConcurrency     int
 	downloadConcurrency   int
-	httpOptions           httpOptions
 	conn                  s3iface.S3API
 }
 
@@ -61,16 +59,11 @@ 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.httpOptions)
+	s.conn, err = createSession(s.aws_access_key_id, s.aws_secret_access_key, s.region, s.endpoint, s.forcePathStyle)
 
 	glog.V(0).Infof("created backend storage s3.%s for region %s bucket %s", s.id, s.region, s.bucket)
 	return
@@ -87,9 +80,6 @@ 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
 }
 
@@ -109,42 +99,6 @@ 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,11 +1,8 @@
 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"
@@ -21,49 +18,12 @@ var (
 	sessionsLock sync.Mutex
 )
 
-// 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) {
+func createSession(awsAccessKeyId, awsSecretAccessKey, region, endpoint string, forcePathStyle bool) (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
 	}
@@ -74,9 +34,6 @@ 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

@@ -1,129 +0,0 @@
-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,25 +61,14 @@ 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 {
-	// 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
+	MasterAddress       pb.ServerAddress
 	grpcDialOption      grpc.DialOption
-	volumeSizeLimit     atomic.Uint64 // read from the master
-	preallocate         atomic.Bool   // read from the master
+	volumeSizeLimit     uint64      // read from the master
+	preallocate         atomic.Bool // read from the master
 	Ip                  string
 	Port                int
 	GrpcPort            int
@@ -880,10 +869,7 @@ 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 {
-		v.noWriteLock.RLock()
-		noWriteOrDelete := v.noWriteOrDelete
-		v.noWriteLock.RUnlock()
-		if noWriteOrDelete {
+		if v.noWriteOrDelete {
 			return 0, fmt.Errorf("volume %d is read only", i)
 		}
 		return v.deleteNeedle2(n)
@@ -989,10 +975,10 @@ func (s *Store) MarkVolumeWritable(i needle.VolumeId) error {
 	prevNoWriteOrDelete := v.noWriteOrDelete
 	prevNoWriteCanDelete := v.noWriteCanDelete
 	v.noWriteOrDelete = 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()
+	// Remote-tiered volumes must stay noWriteCanDelete regardless of marks.
+	if !v.HasRemoteFile() {
+		v.noWriteCanDelete = false
+	}
 	persistErr := v.PersistReadOnly(false, false)
 	if persistErr != nil {
 		var ndErr *volume_info.NotCrashDurableError
@@ -1316,11 +1302,11 @@ func (s *Store) ConfigureVolume(i needle.VolumeId, replication string) error {
 }
 
 func (s *Store) SetVolumeSizeLimit(x uint64) {
-	s.volumeSizeLimit.Store(x)
+	atomic.StoreUint64(&s.volumeSizeLimit, x)
 }
 
 func (s *Store) GetVolumeSizeLimit() uint64 {
-	return s.volumeSizeLimit.Load()
+	return atomic.LoadUint64(&s.volumeSizeLimit)
 }
 
 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{
-		MasterFn:       func() pb.ServerAddress { return masterAddr },
+		MasterAddress:  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{
-		MasterFn:       func() pb.ServerAddress { return masterAddr },
+		MasterAddress:  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{
-		MasterFn:       func() pb.ServerAddress { return masterAddr },
+		MasterAddress:  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{
-		MasterFn:       func() pb.ServerAddress { return masterAddr },
+		MasterAddress:  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
-			logRemoteSuperBlockErr(v.Id, err)
+			glog.Errorf("readSuperBlock remote volume %d: %v", v.Id, err)
 			err = nil
 		}
 	} else {
@@ -434,11 +434,3 @@ 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,33 +68,6 @@ 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

@@ -1,21 +0,0 @@
-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,41 +87,6 @@ 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

@@ -1,123 +0,0 @@
-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)
-}