Merge branch 'master' of https://github.com/seaweedfs/seaweedfs

go.mod

@@ -39,7 +39,7 @@ require (
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/hashicorp/go-uuid v1.0.3 // indirect
-	github.com/jackc/pgx/v5 v5.9.0
+	github.com/jackc/pgx/v5 v5.9.2
 	github.com/jcmturner/gofork v1.7.6 // indirect
 	github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect
 	github.com/jinzhu/copier v0.4.0

go.sum

@@ -1403,8 +1403,8 @@ github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsI
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
 github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
 github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
-github.com/jackc/pgx/v5 v5.9.0 h1:T/dI+2TvmI2H8s/KH1/lXIbz1CUFk3gn5oTjr0/mBsE=
-github.com/jackc/pgx/v5 v5.9.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
+github.com/jackc/pgx/v5 v5.9.2 h1:3ZhOzMWnR4yJ+RW1XImIPsD1aNSz4T4fyP7zlQb56hw=
+github.com/jackc/pgx/v5 v5.9.2/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
 github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
 github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/jaegertracing/jaeger v1.47.0 h1:XXxTMO+GxX930gxKWsg90rFr6RswkCRIW0AgWFnTYsg=

note/slides/Makefile

@@ -0,0 +1,31 @@
+MARP    ?= npx @marp-team/marp-cli
+SRC     := seaweedfs-production-setup.md
+PDF     := $(SRC:.md=.pdf)
+HTML    := $(SRC:.md=.html)
+PPTX    := $(SRC:.md=.pptx)
+FLAGS   := --allow-local-files
+
+.PHONY: all pdf html pptx watch clean
+
+all: pdf
+
+pdf: $(PDF)
+
+html: $(HTML)
+
+pptx: $(PPTX)
+
+$(PDF): $(SRC)
+	$(MARP) $(SRC) --pdf $(FLAGS) -o $@
+
+$(HTML): $(SRC)
+	$(MARP) $(SRC) --html $(FLAGS) -o $@
+
+$(PPTX): $(SRC)
+	$(MARP) $(SRC) --pptx $(FLAGS) -o $@
+
+watch:
+	$(MARP) $(SRC) --watch --html $(FLAGS)
+
+clean:
+	rm -f $(PDF) $(HTML) $(PPTX)

note/slides/seaweedfs-production-setup.md

@@ -0,0 +1,118 @@
+---
+marp: true
+theme: default
+paginate: true
+---
+
+# SeaweedFS Production Setup
+
+A walkthrough of the recommended deployment topology
+
+- Storage layer
+- File access layer
+- Backend operations
+- Erasure coding for durability
+
+---
+
+![bg contain](object-store-layout.png)
+
+---
+
+# Storage Layer
+
+**3 Masters + Volume Servers**
+
+- **Masters (run 3)**
+  - Form a Raft quorum; tolerate 1 failure
+  - Track volume locations and cluster topology
+  - Lightweight; not on the data path for reads/writes after lookup
+
+- **Volume Servers (scale out)**
+  - Hold the actual needle data on local disks
+  - Add more servers for capacity and throughput
+  - Heartbeat to masters; clients talk to them directly for I/O
+
+Rule of thumb: start with 3 masters, then grow volume servers with data.
+
+---
+
+# File Access Layer
+
+**Filer + DB for metadata**
+
+- Filer serves POSIX-like directory and file metadata
+- Metadata store options: PostgreSQL, MySQL, Redis, Cassandra, TiKV, etc.
+- Run **3 filers** for HA
+- Filers are stateless relative to each other — state lives in the DB
+
+**S3 Servers**
+
+- Expose the S3 API on top of the filer
+- Scale the count based on throughput requirements
+- Can be co-located with filers or run independently
+
+---
+
+# Backend Operations
+
+**1 Admin Server + a few Workers**
+
+- **Admin server (1 is enough)**
+  - Coordinates background jobs: balancing, EC encoding, vacuum, replication fixes
+  - **Not on the data path** — a restart does not affect object store reads/writes
+  - No HA required for normal operation
+
+- **Workers (a few)**
+  - Execute long-running tasks issued by the admin server
+  - Horizontally scalable based on maintenance workload
+
+Keeps housekeeping off the hot path.
+
+---
+
+# Production Topology Summary
+
+| Component      | Count                | Role                              |
+|----------------|----------------------|-----------------------------------|
+| Master         | 3                    | Raft quorum, topology, volume map |
+| Volume Server  | N (scale out)        | Stores data needles               |
+| Filer          | 3                    | Metadata gateway                  |
+| Metadata DB    | HA cluster           | Persists filer metadata           |
+| S3 Server      | N (by throughput)    | S3 API frontend                   |
+| Admin          | 1                    | Background job coordinator        |
+| Worker         | A few                | Executes admin-scheduled tasks    |
+
+---
+
+# Erasure Coding for Durability
+
+Prefer **many volume servers** so EC shards spread across failure domains.
+
+**Example: 8 volume servers, EC 5+3**
+
+- Each volume becomes 5 data shards + 3 parity shards
+- Shards are placed on 8 distinct volume servers
+- **Tolerates up to 3 simultaneous volume server failures**
+- Storage overhead: 1.6x vs 3x for full replication
+
+```
+Volume -> [D1][D2][D3][D4][D5] + [P1][P2][P3]
+           |   |   |   |   |     |   |   |
+          VS1 VS2 VS3 VS4 VS5  VS6 VS7 VS8
+```
+
+More volume servers = more EC layouts available (e.g. 10+4, 6+3).
+
+---
+
+# Getting Started Checklist
+
+1. Provision 3 master nodes (small boxes are fine)
+2. Provision volume servers sized to your data footprint
+3. Stand up an HA metadata DB, then 3 filers pointing at it
+4. Add S3 servers sized to required throughput
+5. Run 1 admin + a few workers for background maintenance
+6. Once you have enough volume servers, enable EC (e.g. 5+3 on 8 servers)
+
+Scale each layer independently as usage grows.

seaweed-volume/src/server/grpc_server.rs

@@ -1425,6 +1425,16 @@ impl VolumeServer for VolumeGrpcService {
                         // Determine file path
                         let path = if info.is_ec_volume {
                             let store = self.state.store.read().unwrap();
+                            // std::fs::File::create truncates in place; a mounted
+                            // EcVolume holds fds on the same inodes, so overwriting
+                            // corrupts live readers.
+                            if store.has_ec_volume(VolumeId(info.volume_id)) {
+                                resp_error = Some(format!(
+                                    "ec volume {} is mounted; unmount before ReceiveFile",
+                                    info.volume_id
+                                ));
+                                break;
+                            }
                             // disk_id=0 means "unset" (protobuf default), so auto-select
                             // mirrors VolumeEcShardsCopy: prefer a disk already holding
                             // this volume's shards, then any HDD, then any disk.

test/volume_server/grpc/ec_receive_truncates_mounted_test.go

@@ -0,0 +1,186 @@
+package volume_server_grpc_test
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/test/volume_server/framework"
+	"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+)
+
+func TestReceiveFileRejectsOverwriteOfMountedEcShard(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	clusterHarness := framework.StartVolumeCluster(t, matrix.P1())
+	conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress())
+	defer conn.Close()
+
+	const volumeID = uint32(91841)
+	framework.AllocateVolume(t, grpcClient, volumeID, "")
+
+	httpClient := framework.NewHTTPClient()
+	fid := framework.NewFileID(volumeID, 918401, 0x9184CAFE)
+	uploadResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid,
+		[]byte("ec-receive-overwrite-content-for-issue-9184-repro"))
+	_ = framework.ReadAllAndClose(t, uploadResp)
+	if uploadResp.StatusCode != http.StatusCreated {
+		t.Fatalf("upload expected 201, got %d", uploadResp.StatusCode)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	if _, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{
+		VolumeId:   volumeID,
+		Collection: "",
+	}); err != nil {
+		t.Fatalf("VolumeEcShardsGenerate: %v", err)
+	}
+
+	if _, err := grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{
+		VolumeId:   volumeID,
+		Collection: "",
+		ShardIds:   []uint32{0},
+	}); err != nil {
+		t.Fatalf("VolumeEcShardsMount: %v", err)
+	}
+
+	dataDir := filepath.Join(clusterHarness.BaseDir(), "volume")
+	shardPath := filepath.Join(dataDir, fmt.Sprintf("%d.ec00", volumeID))
+	origInfo, err := os.Stat(shardPath)
+	if err != nil {
+		t.Fatalf("stat mounted shard %s: %v", shardPath, err)
+	}
+	origSize := origInfo.Size()
+	if origSize == 0 {
+		t.Fatalf("mounted shard %s unexpectedly empty", shardPath)
+	}
+
+	readStream, err := grpcClient.VolumeEcShardRead(ctx, &volume_server_pb.VolumeEcShardReadRequest{
+		VolumeId: volumeID,
+		ShardId:  0,
+		Offset:   0,
+		Size:     1,
+	})
+	if err != nil {
+		t.Fatalf("VolumeEcShardRead (pre): %v", err)
+	}
+	if _, err := readStream.Recv(); err != nil {
+		t.Fatalf("VolumeEcShardRead Recv (pre): %v", err)
+	}
+
+	overwritePayload := []byte("bug-9184-overwrite")
+	receiveStream, err := grpcClient.ReceiveFile(ctx)
+	if err != nil {
+		t.Fatalf("ReceiveFile stream create: %v", err)
+	}
+	if err = receiveStream.Send(&volume_server_pb.ReceiveFileRequest{
+		Data: &volume_server_pb.ReceiveFileRequest_Info{
+			Info: &volume_server_pb.ReceiveFileInfo{
+				VolumeId:   volumeID,
+				Ext:        ".ec00",
+				Collection: "",
+				IsEcVolume: true,
+				ShardId:    0,
+				FileSize:   uint64(len(overwritePayload)),
+			},
+		},
+	}); err != nil {
+		t.Fatalf("ReceiveFile send info: %v", err)
+	}
+	resp, err := receiveStream.CloseAndRecv()
+	if err != nil {
+		t.Logf("ReceiveFile rejected at stream level: %v", err)
+	} else {
+		if resp.GetError() == "" {
+			t.Fatalf("expected ReceiveFile to reject overwrite of mounted shard, got success: %+v", resp)
+		}
+		if !strings.Contains(resp.GetError(), "mounted") {
+			t.Fatalf("expected error to mention mounted; got: %s", resp.GetError())
+		}
+	}
+
+	afterInfo, err := os.Stat(shardPath)
+	if err != nil {
+		t.Fatalf("stat shard after rejected overwrite: %v", err)
+	}
+	if afterInfo.Size() != origSize {
+		t.Fatalf("shard %s was modified despite rejection: size was %d, now %d",
+			shardPath, origSize, afterInfo.Size())
+	}
+
+	postStream, err := grpcClient.VolumeEcShardRead(ctx, &volume_server_pb.VolumeEcShardReadRequest{
+		VolumeId: volumeID,
+		ShardId:  0,
+		Offset:   0,
+		Size:     1,
+	})
+	if err != nil {
+		t.Fatalf("VolumeEcShardRead (post): %v", err)
+	}
+	if _, err := postStream.Recv(); err != nil {
+		t.Fatalf("VolumeEcShardRead Recv (post): %v", err)
+	}
+
+	t.Logf("ReceiveFile correctly refused overwrite; mounted shard intact at %d bytes", afterInfo.Size())
+}
+
+func TestReceiveFileAllowsEcShardWhenNoMount(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	clusterHarness := framework.StartVolumeCluster(t, matrix.P1())
+	conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress())
+	defer conn.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	const volumeID = uint32(91843)
+	const collection = "ec-receive-no-mount"
+	payload := []byte("ok-to-receive-not-mounted")
+
+	stream, err := grpcClient.ReceiveFile(ctx)
+	if err != nil {
+		t.Fatalf("ReceiveFile stream create: %v", err)
+	}
+	if err = stream.Send(&volume_server_pb.ReceiveFileRequest{
+		Data: &volume_server_pb.ReceiveFileRequest_Info{
+			Info: &volume_server_pb.ReceiveFileInfo{
+				VolumeId:   volumeID,
+				Ext:        ".ec00",
+				Collection: collection,
+				IsEcVolume: true,
+				ShardId:    0,
+				FileSize:   uint64(len(payload)),
+			},
+		},
+	}); err != nil {
+		t.Fatalf("ReceiveFile send info: %v", err)
+	}
+	if err = stream.Send(&volume_server_pb.ReceiveFileRequest{
+		Data: &volume_server_pb.ReceiveFileRequest_FileContent{FileContent: payload},
+	}); err != nil {
+		t.Fatalf("ReceiveFile send content: %v", err)
+	}
+	resp, err := stream.CloseAndRecv()
+	if err != nil {
+		t.Fatalf("ReceiveFile close: %v", err)
+	}
+	if resp.GetError() != "" {
+		t.Fatalf("expected success on unmounted volume, got error: %s", resp.GetError())
+	}
+	if resp.GetBytesWritten() != uint64(len(payload)) {
+		t.Fatalf("bytes_written mismatch: got %d want %d", resp.GetBytesWritten(), len(payload))
+	}
+}

weed/admin/dash/user_management.go

@@ -247,21 +247,29 @@ func (s *AdminServer) CreateAccessKey(username string, req *CreateAccessKeyReque
 		req = &CreateAccessKeyRequest{}
 	}
 
-	// Validate provided keys
-	if req.AccessKey != "" && (len(req.AccessKey) < 4 || len(req.AccessKey) > 128) {
-		return nil, fmt.Errorf("access key must be between 4 and 128 characters: %w", ErrInvalidInput)
+	// Validate provided keys (shared with the IAM API and embedded IAM paths).
+	if req.AccessKey != "" {
+		if err := iam.ValidateCallerSuppliedAccessKeyId(req.AccessKey); err != nil {
+			return nil, fmt.Errorf("%s: %w", err.Error(), ErrInvalidInput)
+		}
 	}
-	if req.SecretKey != "" && (len(req.SecretKey) < 8 || len(req.SecretKey) > 128) {
-		return nil, fmt.Errorf("secret key must be between 8 and 128 characters: %w", ErrInvalidInput)
+	if req.SecretKey != "" {
+		if err := iam.ValidateCallerSuppliedSecretAccessKey(req.SecretKey); err != nil {
+			return nil, fmt.Errorf("%s: %w", err.Error(), ErrInvalidInput)
+		}
+	}
+	// Enforce the both-or-none rule to match the IAM API and embedded IAM
+	// paths — silently generating the missing half lets a caller end up
+	// with a credential they did not fully choose.
+	if (req.AccessKey != "") != (req.SecretKey != "") {
+		return nil, fmt.Errorf("access key and secret key must be supplied together: %w", ErrInvalidInput)
 	}
 
 	// Use provided keys or generate new ones
 	accessKey := req.AccessKey
+	secretKey := req.SecretKey
 	if accessKey == "" {
 		accessKey = generateAccessKey()
-	}
-	secretKey := req.SecretKey
-	if secretKey == "" {
 		secretKey = generateSecretKey()
 	}
 

weed/filer/redis3/kv_directory_children_test.go

@@ -2,109 +2,14 @@ package redis3
 
 import (
 	"context"
-	"fmt"
 	"strconv"
 	"testing"
-	"time"
 
 	"github.com/redis/go-redis/v9"
 	"github.com/stvp/tempredis"
 )
 
-var names = []string{
-	"cassandra.in.sh",
-	"cassandra",
-	"debug-cql.bat",
-	"nodetool",
-	"nodetool.bat",
-	"source-conf.ps1",
-	"sstableloader",
-	"sstableloader.bat",
-	"sstablescrub",
-	"sstablescrub.bat",
-	"sstableupgrade",
-	"sstableupgrade.bat",
-	"sstableutil",
-	"sstableutil.bat",
-	"sstableverify",
-	"sstableverify.bat",
-	"stop-server",
-	"stop-server.bat",
-	"stop-server.ps1",
-	"cassandra.in.bat",
-	"cqlsh.py",
-	"cqlsh",
-	"cassandra.ps1",
-	"cqlsh.bat",
-	"debug-cql",
-	"cassandra.bat",
-}
-
-func yTestNameList(t *testing.T) {
-	server, err := tempredis.Start(tempredis.Config{})
-	if err != nil {
-		panic(err)
-	}
-	defer server.Term()
-
-	client := redis.NewClient(&redis.Options{
-		Network: "unix",
-		Addr:    server.Socket(),
-	})
-
-	store := newSkipListElementStore("/yyy/bin", client)
-	var data []byte
-	for _, name := range names {
-		nameList := LoadItemList(data, "/yyy/bin", client, store, maxNameBatchSizeLimit)
-		nameList.WriteName(name)
-
-		nameList.ListNames("", func(name string) bool {
-			println(name)
-			return true
-		})
-
-		if nameList.HasChanges() {
-			data = nameList.ToBytes()
-		}
-		println()
-	}
-
-	nameList := LoadItemList(data, "/yyy/bin", client, store, maxNameBatchSizeLimit)
-	nameList.ListNames("", func(name string) bool {
-		println(name)
-		return true
-	})
-
-}
-
-func yBenchmarkNameList(b *testing.B) {
-
-	server, err := tempredis.Start(tempredis.Config{})
-	if err != nil {
-		panic(err)
-	}
-	defer server.Term()
-
-	client := redis.NewClient(&redis.Options{
-		Network: "unix",
-		Addr:    server.Socket(),
-	})
-
-	store := newSkipListElementStore("/yyy/bin", client)
-	var data []byte
-	for i := 0; i < b.N; i++ {
-		nameList := LoadItemList(data, "/yyy/bin", client, store, maxNameBatchSizeLimit)
-
-		nameList.WriteName(strconv.Itoa(i) + "namexxxxxxxxxxxxxxxxxxx")
-
-		if nameList.HasChanges() {
-			data = nameList.ToBytes()
-		}
-	}
-}
-
 func BenchmarkRedis(b *testing.B) {
-
 	server, err := tempredis.Start(tempredis.Config{})
 	if err != nil {
 		panic(err)
@@ -120,92 +25,3 @@ func BenchmarkRedis(b *testing.B) {
 		client.ZAddNX(context.Background(), "/yyy/bin", redis.Z{Score: 0, Member: strconv.Itoa(i) + "namexxxxxxxxxxxxxxxxxxx"})
 	}
 }
-
-func xTestNameListAdd(t *testing.T) {
-
-	server, err := tempredis.Start(tempredis.Config{})
-	if err != nil {
-		panic(err)
-	}
-	defer server.Term()
-
-	client := redis.NewClient(&redis.Options{
-		Addr:     "localhost:6379",
-		Password: "",
-		DB:       0,
-	})
-
-	client.FlushAll(context.Background())
-
-	N := 364800
-
-	ts0 := time.Now()
-	store := newSkipListElementStore("/y", client)
-	var data []byte
-	nameList := LoadItemList(data, "/y", client, store, 100000)
-	for i := 0; i < N; i++ {
-		nameList.WriteName(fmt.Sprintf("%8d", i))
-	}
-
-	ts1 := time.Now()
-
-	for i := 0; i < N; i++ {
-		client.ZAddNX(context.Background(), "/x", redis.Z{Score: 0, Member: fmt.Sprintf("name %8d", i)})
-	}
-	ts2 := time.Now()
-
-	fmt.Printf("%v %v", ts1.Sub(ts0), ts2.Sub(ts1))
-
-	/*
-		keys := client.Keys(context.Background(), "/*m").Val()
-		for _, k := range keys {
-			println("key", k)
-			for i, v := range client.ZRangeByLex(context.Background(), k, &redis.ZRangeBy{
-				Min:    "-",
-				Max:    "+",
-			}).Val() {
-				println(" ", i, v)
-			}
-		}
-	*/
-}
-
-func xBenchmarkNameList(b *testing.B) {
-
-	server, err := tempredis.Start(tempredis.Config{})
-	if err != nil {
-		panic(err)
-	}
-	defer server.Term()
-
-	client := redis.NewClient(&redis.Options{
-		Addr:     "localhost:6379",
-		Password: "",
-		DB:       0,
-	})
-
-	store := newSkipListElementStore("/yyy/bin", client)
-	var data []byte
-	for i := 0; i < b.N; i++ {
-		nameList := LoadItemList(data, "/yyy/bin", client, store, maxNameBatchSizeLimit)
-
-		nameList.WriteName(fmt.Sprintf("name %8d", i))
-
-		if nameList.HasChanges() {
-			data = nameList.ToBytes()
-		}
-	}
-}
-
-func xBenchmarkRedis(b *testing.B) {
-
-	client := redis.NewClient(&redis.Options{
-		Addr:     "localhost:6379",
-		Password: "",
-		DB:       0,
-	})
-
-	for i := 0; i < b.N; i++ {
-		client.ZAddNX(context.Background(), "/xxx/bin", redis.Z{Score: 0, Member: fmt.Sprintf("name %8d", i)})
-	}
-}

weed/iam/redact.go

@@ -0,0 +1,32 @@
+package iam
+
+import "net/url"
+
+// sensitiveFormKeys is the set of IAM request form parameters whose values
+// must never be written to logs. Matching is case-sensitive and uses the
+// exact AWS IAM parameter name. Extend this when adding IAM actions that
+// accept credentials, passwords, session tokens, or private keys.
+var sensitiveFormKeys = map[string]struct{}{
+	"SecretAccessKey": {},
+	"Password":        {},
+	"NewPassword":     {},
+	"OldPassword":     {},
+	"PrivateKey":      {},
+	"SessionToken":    {},
+}
+
+// RedactSensitiveFormValues returns a shallow copy of values with every
+// sensitive key (see sensitiveFormKeys) replaced by "[REDACTED]". Intended
+// for debug-level logging of IAM request forms so secrets do not leak into
+// log sinks.
+func RedactSensitiveFormValues(values url.Values) url.Values {
+	safe := make(url.Values, len(values))
+	for k, v := range values {
+		if _, sensitive := sensitiveFormKeys[k]; sensitive {
+			safe[k] = []string{"[REDACTED]"}
+		} else {
+			safe[k] = v
+		}
+	}
+	return safe
+}

weed/iam/redact_test.go

@@ -0,0 +1,33 @@
+package iam
+
+import (
+	"net/url"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestRedactSensitiveFormValues(t *testing.T) {
+	in := url.Values{
+		"Action":          []string{"CreateAccessKey"},
+		"UserName":        []string{"alice"},
+		"AccessKeyId":     []string{"AKIAALICE1"},
+		"SecretAccessKey": []string{"very-secret"},
+		"Password":        []string{"hunter2"},
+		"NewPassword":     []string{"new-hunter2"},
+		"OldPassword":     []string{"old-hunter2"},
+		"PrivateKey":      []string{"-----BEGIN-----"},
+		"SessionToken":    []string{"tok"},
+	}
+	out := RedactSensitiveFormValues(in)
+
+	assert.Equal(t, []string{"CreateAccessKey"}, out["Action"])
+	assert.Equal(t, []string{"alice"}, out["UserName"])
+	assert.Equal(t, []string{"AKIAALICE1"}, out["AccessKeyId"])
+	for _, k := range []string{"SecretAccessKey", "Password", "NewPassword", "OldPassword", "PrivateKey", "SessionToken"} {
+		assert.Equal(t, []string{"[REDACTED]"}, out[k], "key %q should be redacted", k)
+	}
+
+	// Input is not mutated.
+	assert.Equal(t, []string{"very-secret"}, in["SecretAccessKey"])
+}

weed/iam/validation.go

@@ -0,0 +1,68 @@
+package iam
+
+import (
+	"fmt"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+)
+
+// ValidateCallerSuppliedAccessKeyId checks that a caller-supplied AccessKeyId
+// is 4 to 128 ASCII alphanumeric characters. Returns nil if valid.
+//
+// The alphanumeric restriction avoids characters that would break SigV4
+// canonicalization (e.g. '/' and '=' appear as delimiters in Credential
+// headers), so this is a stricter superset of the rule AWS enforces.
+func ValidateCallerSuppliedAccessKeyId(accessKeyId string) error {
+	if len(accessKeyId) < 4 || len(accessKeyId) > 128 {
+		return fmt.Errorf("AccessKeyId must be 4 to 128 alphanumeric characters")
+	}
+	for _, r := range accessKeyId {
+		if !((r >= 'A' && r <= 'Z') || (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9')) {
+			return fmt.Errorf("AccessKeyId must be 4 to 128 alphanumeric characters")
+		}
+	}
+	return nil
+}
+
+// ValidateCallerSuppliedSecretAccessKey checks that a caller-supplied
+// SecretAccessKey is 8 to 128 characters. Returns nil if valid.
+func ValidateCallerSuppliedSecretAccessKey(secretAccessKey string) error {
+	if len(secretAccessKey) < 8 || len(secretAccessKey) > 128 {
+		return fmt.Errorf("SecretAccessKey must be between 8 and 128 characters")
+	}
+	return nil
+}
+
+// AccessKeyOwner identifies which entity in an S3ApiConfiguration already owns
+// a given AccessKeyId. Returned by FindAccessKeyOwner for collision checks on
+// caller-supplied credentials.
+type AccessKeyOwner struct {
+	// Type is "user" or "service account".
+	Type string
+	// Name is the identity's Name (for users) or the service account's Id.
+	Name string
+}
+
+// FindAccessKeyOwner scans s3cfg for an identity or service account whose
+// credentials already contain accessKeyId. Returns nil if the key is free.
+//
+// Callers should log Name only at debug level — error responses returned to
+// the caller should not include owner identity to avoid information leaks.
+func FindAccessKeyOwner(s3cfg *iam_pb.S3ApiConfiguration, accessKeyId string) *AccessKeyOwner {
+	if s3cfg == nil || accessKeyId == "" {
+		return nil
+	}
+	for _, ident := range s3cfg.Identities {
+		for _, cred := range ident.Credentials {
+			if cred.AccessKey == accessKeyId {
+				return &AccessKeyOwner{Type: "user", Name: ident.Name}
+			}
+		}
+	}
+	for _, sa := range s3cfg.ServiceAccounts {
+		if sa.Credential != nil && sa.Credential.AccessKey == accessKeyId {
+			return &AccessKeyOwner{Type: "service account", Name: sa.Id}
+		}
+	}
+	return nil
+}

weed/iam/validation_test.go

@@ -0,0 +1,104 @@
+package iam
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestValidateCallerSuppliedAccessKeyId(t *testing.T) {
+	cases := []struct {
+		name    string
+		input   string
+		wantErr bool
+	}{
+		{"empty", "", true},
+		{"three chars", "abc", true},
+		{"four chars ok", "abcd", false},
+		{"mixed case alnum", "MyAppKey123", false},
+		{"128 chars ok", strings.Repeat("a", 128), false},
+		{"129 chars too long", strings.Repeat("a", 129), true},
+		{"slash rejected", "foo/bar", true},
+		{"equals rejected", "foo=bar", true},
+		{"dash rejected", "foo-bar", true},
+		{"underscore rejected", "foo_bar", true},
+		{"unicode rejected", "fooö123", true},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			err := ValidateCallerSuppliedAccessKeyId(tc.input)
+			if tc.wantErr {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}
+
+func TestValidateCallerSuppliedSecretAccessKey(t *testing.T) {
+	cases := []struct {
+		name    string
+		input   string
+		wantErr bool
+	}{
+		{"empty", "", true},
+		{"seven chars", "abcdefg", true},
+		{"eight chars ok", "abcdefgh", false},
+		{"128 chars ok", strings.Repeat("a", 128), false},
+		{"129 chars too long", strings.Repeat("a", 129), true},
+		{"non-alnum allowed in secret", "sec/ret=1", false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			err := ValidateCallerSuppliedSecretAccessKey(tc.input)
+			if tc.wantErr {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}
+
+func TestFindAccessKeyOwner(t *testing.T) {
+	s3cfg := &iam_pb.S3ApiConfiguration{
+		Identities: []*iam_pb.Identity{
+			{
+				Name: "alice",
+				Credentials: []*iam_pb.Credential{
+					{AccessKey: "AKIAALICE1", SecretKey: "s"},
+				},
+			},
+			{Name: "bob"},
+		},
+		ServiceAccounts: []*iam_pb.ServiceAccount{
+			{Id: "svc-1", Credential: &iam_pb.Credential{AccessKey: "SVCKEY1"}},
+			{Id: "svc-2"},
+		},
+	}
+
+	t.Run("matches user", func(t *testing.T) {
+		owner := FindAccessKeyOwner(s3cfg, "AKIAALICE1")
+		assert.NotNil(t, owner)
+		assert.Equal(t, "user", owner.Type)
+		assert.Equal(t, "alice", owner.Name)
+	})
+	t.Run("matches service account", func(t *testing.T) {
+		owner := FindAccessKeyOwner(s3cfg, "SVCKEY1")
+		assert.NotNil(t, owner)
+		assert.Equal(t, "service account", owner.Type)
+		assert.Equal(t, "svc-1", owner.Name)
+	})
+	t.Run("no match", func(t *testing.T) {
+		assert.Nil(t, FindAccessKeyOwner(s3cfg, "NOTTAKEN"))
+	})
+	t.Run("empty key", func(t *testing.T) {
+		assert.Nil(t, FindAccessKeyOwner(s3cfg, ""))
+	})
+	t.Run("nil config", func(t *testing.T) {
+		assert.Nil(t, FindAccessKeyOwner(nil, "anything"))
+	})
+}

weed/iamapi/iamapi_management_handlers.go

@@ -929,13 +929,38 @@ func (iama *IamApiServer) CreateAccessKey(s3cfg *iam_pb.S3ApiConfiguration, valu
 	userName := values.Get("UserName")
 	status := iam.StatusTypeActive
 
-	accessKeyId, err := StringWithCharset(21, charsetUpper)
-	if err != nil {
-		return resp, &IamError{Code: iam.ErrCodeServiceFailureException, Error: fmt.Errorf("failed to generate access key: %w", err)}
+	accessKeyId := values.Get("AccessKeyId")
+	secretAccessKey := values.Get("SecretAccessKey")
+	if accessKeyId != "" {
+		if err := iamlib.ValidateCallerSuppliedAccessKeyId(accessKeyId); err != nil {
+			return resp, &IamError{Code: iam.ErrCodeInvalidInputException, Error: err}
+		}
 	}
-	secretAccessKey, err := StringWithCharset(42, charset)
-	if err != nil {
-		return resp, &IamError{Code: iam.ErrCodeServiceFailureException, Error: fmt.Errorf("failed to generate secret key: %w", err)}
+	if secretAccessKey != "" {
+		if err := iamlib.ValidateCallerSuppliedSecretAccessKey(secretAccessKey); err != nil {
+			return resp, &IamError{Code: iam.ErrCodeInvalidInputException, Error: err}
+		}
+	}
+	if (accessKeyId != "") != (secretAccessKey != "") {
+		return resp, &IamError{Code: iam.ErrCodeInvalidInputException, Error: fmt.Errorf("AccessKeyId and SecretAccessKey must be supplied together")}
+	}
+	if owner := iamlib.FindAccessKeyOwner(s3cfg, accessKeyId); owner != nil {
+		glog.V(4).Infof("CreateAccessKey: supplied AccessKeyId already in use by %s %s", owner.Type, owner.Name)
+		return resp, &IamError{Code: iam.ErrCodeEntityAlreadyExistsException, Error: fmt.Errorf("AccessKeyId is already in use")}
+	}
+	if accessKeyId == "" {
+		var err error
+		accessKeyId, err = StringWithCharset(21, charsetUpper)
+		if err != nil {
+			return resp, &IamError{Code: iam.ErrCodeServiceFailureException, Error: fmt.Errorf("failed to generate access key: %w", err)}
+		}
+	}
+	if secretAccessKey == "" {
+		var err error
+		secretAccessKey, err = StringWithCharset(42, charset)
+		if err != nil {
+			return resp, &IamError{Code: iam.ErrCodeServiceFailureException, Error: fmt.Errorf("failed to generate secret key: %w", err)}
+		}
 	}
 
 	resp.CreateAccessKeyResult.AccessKey.AccessKeyId = &accessKeyId
@@ -1085,7 +1110,7 @@ func (iama *IamApiServer) DoActions(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	glog.V(4).Infof("DoActions: %+v", values)
+	glog.V(4).Infof("DoActions: %+v", iamlib.RedactSensitiveFormValues(values))
 	var response iamlib.RequestIDSetter
 	changed := true
 	switch r.Form.Get("Action") {

weed/iamapi/iamapi_management_handlers_test.go

@@ -3,6 +3,7 @@ package iamapi
 import (
 	"encoding/json"
 	"net/url"
+	"strings"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/service/iam"
@@ -629,3 +630,266 @@ func TestListAttachedUserPolicies(t *testing.T) {
 	assert.NotNil(t, iamErr)
 	assert.Equal(t, iam.ErrCodeNoSuchEntityException, iamErr.Code)
 }
+
+func TestCreateAccessKeyWithCallerSuppliedKeys(t *testing.T) {
+	iama := newTestIamApiServer(Policies{})
+	s3cfg := &iam_pb.S3ApiConfiguration{
+		Identities: []*iam_pb.Identity{{Name: "alice"}},
+	}
+	values := url.Values{
+		"UserName":       []string{"alice"},
+		"AccessKeyId":    []string{"myappkey"},
+		"SecretAccessKey": []string{"mysecret1234"},
+	}
+	resp, iamErr := iama.CreateAccessKey(s3cfg, values)
+	assert.Nil(t, iamErr)
+	assert.Equal(t, "myappkey", *resp.CreateAccessKeyResult.AccessKey.AccessKeyId)
+	assert.Equal(t, "mysecret1234", *resp.CreateAccessKeyResult.AccessKey.SecretAccessKey)
+	assert.Equal(t, "alice", *resp.CreateAccessKeyResult.AccessKey.UserName)
+	assert.Equal(t, "myappkey", s3cfg.Identities[0].Credentials[0].AccessKey)
+	assert.Equal(t, "mysecret1234", s3cfg.Identities[0].Credentials[0].SecretKey)
+}
+
+func TestCreateAccessKeyRandomGeneration(t *testing.T) {
+	iama := newTestIamApiServer(Policies{})
+	s3cfg := &iam_pb.S3ApiConfiguration{
+		Identities: []*iam_pb.Identity{{Name: "alice"}},
+	}
+	values := url.Values{
+		"UserName": []string{"alice"},
+	}
+	resp, iamErr := iama.CreateAccessKey(s3cfg, values)
+	assert.Nil(t, iamErr)
+	assert.NotEmpty(t, *resp.CreateAccessKeyResult.AccessKey.AccessKeyId)
+	assert.NotEmpty(t, *resp.CreateAccessKeyResult.AccessKey.SecretAccessKey)
+	assert.Len(t, s3cfg.Identities[0].Credentials, 1)
+}
+
+func TestCreateAccessKeyRejectsWeakKeys(t *testing.T) {
+	iama := newTestIamApiServer(Policies{})
+	s3cfg := &iam_pb.S3ApiConfiguration{
+		Identities: []*iam_pb.Identity{{Name: "alice"}},
+	}
+	// Too short
+	values := url.Values{
+		"UserName":       []string{"alice"},
+		"AccessKeyId":    []string{"ab"},
+		"SecretAccessKey": []string{"validsecret1"},
+	}
+	_, iamErr := iama.CreateAccessKey(s3cfg, values)
+	assert.NotNil(t, iamErr)
+	assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code)
+
+	// Short secret
+	values = url.Values{
+		"UserName":       []string{"alice"},
+		"AccessKeyId":    []string{"validkey"},
+		"SecretAccessKey": []string{"short"},
+	}
+	_, iamErr = iama.CreateAccessKey(s3cfg, values)
+	assert.NotNil(t, iamErr)
+	assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code)
+
+	// SigV4 delimiters
+	values = url.Values{
+		"UserName":       []string{"alice"},
+		"AccessKeyId":    []string{"foo/bar=baz"},
+		"SecretAccessKey": []string{"validsecret1"},
+	}
+	_, iamErr = iama.CreateAccessKey(s3cfg, values)
+	assert.NotNil(t, iamErr)
+	assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code)
+}
+
+func TestCreateAccessKeyRejectsCollision(t *testing.T) {
+	iama := newTestIamApiServer(Policies{})
+	// Use a distinctive owner name ("ownerAlpha") that shares no substring
+	// with the expected error message so the leak assertion is meaningful.
+	const ownerName = "ownerAlpha"
+
+	t.Run("identity credential", func(t *testing.T) {
+		s3cfg := &iam_pb.S3ApiConfiguration{
+			Identities: []*iam_pb.Identity{
+				{
+					Name: ownerName,
+					Credentials: []*iam_pb.Credential{
+						{AccessKey: "takenkey", SecretKey: "existingsecret"},
+					},
+				},
+				{Name: "newuser"},
+			},
+		}
+		values := url.Values{
+			"UserName":        []string{"newuser"},
+			"AccessKeyId":     []string{"takenkey"},
+			"SecretAccessKey": []string{"newsecret123"},
+		}
+		_, iamErr := iama.CreateAccessKey(s3cfg, values)
+		assert.NotNil(t, iamErr)
+		assert.Equal(t, iam.ErrCodeEntityAlreadyExistsException, iamErr.Code)
+		assert.NotContains(t, iamErr.Error.Error(), ownerName, "should not leak owner name")
+		assert.Len(t, s3cfg.Identities[1].Credentials, 0)
+	})
+
+	t.Run("service account credential", func(t *testing.T) {
+		const saId = "svcAlpha"
+		s3cfg := &iam_pb.S3ApiConfiguration{
+			Identities: []*iam_pb.Identity{
+				{Name: "newuser"},
+			},
+			ServiceAccounts: []*iam_pb.ServiceAccount{
+				{
+					Id:         saId,
+					Credential: &iam_pb.Credential{AccessKey: "takenkey", SecretKey: "existingsecret"},
+				},
+			},
+		}
+		values := url.Values{
+			"UserName":        []string{"newuser"},
+			"AccessKeyId":     []string{"takenkey"},
+			"SecretAccessKey": []string{"newsecret123"},
+		}
+		_, iamErr := iama.CreateAccessKey(s3cfg, values)
+		assert.NotNil(t, iamErr)
+		assert.Equal(t, iam.ErrCodeEntityAlreadyExistsException, iamErr.Code)
+		assert.NotContains(t, iamErr.Error.Error(), saId, "should not leak owner id")
+		// The service account's existing credential must be untouched, and
+		// no new credential should be attached to the identity.
+		assert.Equal(t, "takenkey", s3cfg.ServiceAccounts[0].Credential.AccessKey)
+		assert.Equal(t, "existingsecret", s3cfg.ServiceAccounts[0].Credential.SecretKey)
+		assert.Len(t, s3cfg.Identities[0].Credentials, 0)
+	})
+}
+
+func TestCreateAccessKeyBoundary(t *testing.T) {
+	iama := newTestIamApiServer(Policies{})
+	s3cfg := &iam_pb.S3ApiConfiguration{
+		Identities: []*iam_pb.Identity{{Name: "alice"}},
+	}
+	// Exactly 4 chars - should pass
+	values := url.Values{
+		"UserName":        []string{"alice"},
+		"AccessKeyId":     []string{"abcd"},
+		"SecretAccessKey": []string{"secretkey123"},
+	}
+	resp, iamErr := iama.CreateAccessKey(s3cfg, values)
+	assert.Nil(t, iamErr)
+	assert.Equal(t, "abcd", *resp.CreateAccessKeyResult.AccessKey.AccessKeyId)
+
+	// Exactly 3 chars - should fail
+	s3cfg.Identities[0].Credentials = nil
+	values = url.Values{
+		"UserName":        []string{"alice"},
+		"AccessKeyId":     []string{"abc"},
+		"SecretAccessKey": []string{"secretkey123"},
+	}
+	_, iamErr = iama.CreateAccessKey(s3cfg, values)
+	assert.NotNil(t, iamErr)
+	assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code)
+
+	// Exactly 128 chars - should pass
+	s3cfg.Identities[0].Credentials = nil
+	ak128 := strings.Repeat("a", 128)
+	sk128 := strings.Repeat("s", 128)
+	values = url.Values{
+		"UserName":        []string{"alice"},
+		"AccessKeyId":     []string{ak128},
+		"SecretAccessKey": []string{sk128},
+	}
+	resp, iamErr = iama.CreateAccessKey(s3cfg, values)
+	assert.Nil(t, iamErr)
+	assert.Equal(t, ak128, *resp.CreateAccessKeyResult.AccessKey.AccessKeyId)
+	assert.Equal(t, sk128, *resp.CreateAccessKeyResult.AccessKey.SecretAccessKey)
+
+	// 129 chars AccessKeyId - should fail
+	s3cfg.Identities[0].Credentials = nil
+	values = url.Values{
+		"UserName":        []string{"alice"},
+		"AccessKeyId":     []string{strings.Repeat("a", 129)},
+		"SecretAccessKey": []string{sk128},
+	}
+	_, iamErr = iama.CreateAccessKey(s3cfg, values)
+	assert.NotNil(t, iamErr)
+	assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code)
+
+	// 7-char SecretAccessKey - should fail
+	s3cfg.Identities[0].Credentials = nil
+	values = url.Values{
+		"UserName":        []string{"alice"},
+		"AccessKeyId":     []string{"validkey"},
+		"SecretAccessKey": []string{"1234567"},
+	}
+	_, iamErr = iama.CreateAccessKey(s3cfg, values)
+	assert.NotNil(t, iamErr)
+	assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code)
+
+	// Exactly 8-char SecretAccessKey - should pass (lower boundary)
+	s3cfg.Identities[0].Credentials = nil
+	values = url.Values{
+		"UserName":        []string{"alice"},
+		"AccessKeyId":     []string{"validkey"},
+		"SecretAccessKey": []string{"12345678"},
+	}
+	resp, iamErr = iama.CreateAccessKey(s3cfg, values)
+	assert.Nil(t, iamErr)
+	assert.Equal(t, "12345678", *resp.CreateAccessKeyResult.AccessKey.SecretAccessKey)
+
+	// 129-char SecretAccessKey - should fail
+	s3cfg.Identities[0].Credentials = nil
+	values = url.Values{
+		"UserName":        []string{"alice"},
+		"AccessKeyId":     []string{"validkey"},
+		"SecretAccessKey": []string{strings.Repeat("s", 129)},
+	}
+	_, iamErr = iama.CreateAccessKey(s3cfg, values)
+	assert.NotNil(t, iamErr)
+	assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code)
+}
+
+func TestCreateAccessKeyRejectsPartialSupply(t *testing.T) {
+	iama := newTestIamApiServer(Policies{})
+	s3cfg := &iam_pb.S3ApiConfiguration{
+		Identities: []*iam_pb.Identity{{Name: "alice"}},
+	}
+	// AccessKeyId supplied, SecretAccessKey omitted
+	values := url.Values{
+		"UserName":    []string{"alice"},
+		"AccessKeyId": []string{"myappkey"},
+	}
+	_, iamErr := iama.CreateAccessKey(s3cfg, values)
+	assert.NotNil(t, iamErr)
+	assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code)
+	assert.Len(t, s3cfg.Identities[0].Credentials, 0)
+
+	// SecretAccessKey supplied, AccessKeyId omitted
+	values = url.Values{
+		"UserName":        []string{"alice"},
+		"SecretAccessKey": []string{"secretkey123"},
+	}
+	_, iamErr = iama.CreateAccessKey(s3cfg, values)
+	assert.NotNil(t, iamErr)
+	assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code)
+	assert.Len(t, s3cfg.Identities[0].Credentials, 0)
+
+	// Partial supply wins over collision: only AccessKeyId supplied, and
+	// it matches an existing credential. We must see InvalidInput, not
+	// EntityAlreadyExists — the both-or-none rule is more fundamental.
+	s3cfg = &iam_pb.S3ApiConfiguration{
+		Identities: []*iam_pb.Identity{
+			{
+				Name: "ownerAlpha",
+				Credentials: []*iam_pb.Credential{
+					{AccessKey: "takenkey", SecretKey: "s"},
+				},
+			},
+			{Name: "alice"},
+		},
+	}
+	values = url.Values{
+		"UserName":    []string{"alice"},
+		"AccessKeyId": []string{"takenkey"},
+	}
+	_, iamErr = iama.CreateAccessKey(s3cfg, values)
+	assert.NotNil(t, iamErr)
+	assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code)
+}

weed/operation/upload_content.go

@@ -455,6 +455,7 @@ func (uploader *Uploader) upload_content(ctx context.Context, fillBufferFunction
 	etag := getEtag(resp)
 	if resp.StatusCode == http.StatusNoContent {
 		ret.ETag = etag
+		ret.ContentMd5 = resp.Header.Get("Content-MD5")
 		return &ret, nil
 	}
 

weed/s3api/s3api_embedded_iam.go

@@ -410,32 +410,61 @@ func (e *EmbeddedIamApi) CreateAccessKey(s3cfg *iam_pb.S3ApiConfiguration, value
 	userName := values.Get("UserName")
 	status := iam.StatusTypeActive
 
-	// Generate AWS-standard access key: AKIA prefix + 16 random uppercase chars = 20 total
-	randomPart, err := iamStringWithCharset(AccessKeyLength-len(UserAccessKeyPrefix), iamCharsetUpper)
-	if err != nil {
-		return resp, &iamError{Code: iam.ErrCodeServiceFailureException, Error: fmt.Errorf("failed to generate access key: %w", err)}
+	accessKeyId := values.Get("AccessKeyId")
+	secretAccessKey := values.Get("SecretAccessKey")
+	if accessKeyId != "" {
+		if err := iamlib.ValidateCallerSuppliedAccessKeyId(accessKeyId); err != nil {
+			return resp, &iamError{Code: iam.ErrCodeInvalidInputException, Error: err}
+		}
 	}
-	accessKeyId := UserAccessKeyPrefix + randomPart
-
-	secretAccessKey, err := iamStringWithCharset(SecretKeyLength, iamCharset)
-	if err != nil {
-		return resp, &iamError{Code: iam.ErrCodeServiceFailureException, Error: fmt.Errorf("failed to generate secret key: %w", err)}
+	if secretAccessKey != "" {
+		if err := iamlib.ValidateCallerSuppliedSecretAccessKey(secretAccessKey); err != nil {
+			return resp, &iamError{Code: iam.ErrCodeInvalidInputException, Error: err}
+		}
+	}
+	if (accessKeyId != "") != (secretAccessKey != "") {
+		return resp, &iamError{Code: iam.ErrCodeInvalidInputException, Error: fmt.Errorf("AccessKeyId and SecretAccessKey must be supplied together")}
 	}
 
+	// Find the target user before touching the RNG or scanning for collisions,
+	// so a missing user fails fast without consuming entropy.
+	var target *iam_pb.Identity
+	for _, ident := range s3cfg.Identities {
+		if userName == ident.Name {
+			target = ident
+			break
+		}
+	}
+	if target == nil {
+		return resp, &iamError{Code: iam.ErrCodeNoSuchEntityException, Error: fmt.Errorf(iamUserDoesNotExist, userName)}
+	}
+
+	if owner := iamlib.FindAccessKeyOwner(s3cfg, accessKeyId); owner != nil {
+		glog.V(4).Infof("CreateAccessKey: supplied AccessKeyId already in use by %s %s", owner.Type, owner.Name)
+		return resp, &iamError{Code: iam.ErrCodeEntityAlreadyExistsException, Error: fmt.Errorf("AccessKeyId is already in use")}
+	}
+	if accessKeyId == "" {
+		randomPart, err := iamStringWithCharset(AccessKeyLength-len(UserAccessKeyPrefix), iamCharsetUpper)
+		if err != nil {
+			return resp, &iamError{Code: iam.ErrCodeServiceFailureException, Error: fmt.Errorf("failed to generate access key: %w", err)}
+		}
+		accessKeyId = UserAccessKeyPrefix + randomPart
+	}
+	if secretAccessKey == "" {
+		var err error
+		secretAccessKey, err = iamStringWithCharset(SecretKeyLength, iamCharset)
+		if err != nil {
+			return resp, &iamError{Code: iam.ErrCodeServiceFailureException, Error: fmt.Errorf("failed to generate secret key: %w", err)}
+		}
+	}
 	resp.CreateAccessKeyResult.AccessKey.AccessKeyId = &accessKeyId
 	resp.CreateAccessKeyResult.AccessKey.SecretAccessKey = &secretAccessKey
 	resp.CreateAccessKeyResult.AccessKey.UserName = &userName
 	resp.CreateAccessKeyResult.AccessKey.Status = &status
 
-	for _, ident := range s3cfg.Identities {
-		if userName == ident.Name {
-			ident.Credentials = append(ident.Credentials,
-				&iam_pb.Credential{AccessKey: accessKeyId, SecretKey: secretAccessKey, Status: iamAccessKeyStatusActive})
-			return resp, nil
-		}
-	}
-	// User not found - return error instead of implicitly creating the user
-	return resp, &iamError{Code: iam.ErrCodeNoSuchEntityException, Error: fmt.Errorf(iamUserDoesNotExist, userName)}
+	target.Credentials = append(target.Credentials,
+		&iam_pb.Credential{AccessKey: accessKeyId, SecretKey: secretAccessKey, Status: iamAccessKeyStatusActive})
+	return resp, nil
 }
 
 // DeleteAccessKey deletes an access key for a user.
@@ -2102,7 +2131,7 @@ func (e *EmbeddedIamApi) ExecuteAction(ctx context.Context, values url.Values, s
 		return nil, &iamError{Code: s3err.GetAPIError(s3err.ErrInternalError).Code, Error: fmt.Errorf("failed to get s3 api configuration: %v", err)}
 	}
 
-	glog.V(4).Infof("IAM ExecuteAction: %+v", values)
+	glog.V(4).Infof("IAM ExecuteAction: %+v", iamlib.RedactSensitiveFormValues(values))
 	var response iamlib.RequestIDSetter
 	changed := true
 	switch values.Get("Action") {

weed/s3api/s3api_embedded_iam_test.go

@@ -911,6 +911,356 @@ func TestEmbeddedIamCreateAccessKey(t *testing.T) {
 	assert.Len(t, api.mockConfig.Identities[0].Credentials, 1)
 }
 
+// TestEmbeddedIamCreateAccessKeyRejectsMissingUser verifies CreateAccessKey
+// returns NoSuchEntity for an unknown user without mutating the config.
+func TestEmbeddedIamCreateAccessKeyRejectsMissingUser(t *testing.T) {
+	api := NewEmbeddedIamApiForTest()
+	api.mockConfig = &iam_pb.S3ApiConfiguration{
+		Identities: []*iam_pb.Identity{{Name: "ExistingUser"}},
+	}
+
+	form := url.Values{}
+	form.Set("Action", "CreateAccessKey")
+	form.Set("UserName", "GhostUser")
+
+	req, _ := http.NewRequest("POST", "/", nil)
+	req.PostForm = form
+	req.Form = form
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	rr := httptest.NewRecorder()
+	apiRouter := mux.NewRouter().SkipClean(true)
+	apiRouter.Path("/").Methods(http.MethodPost).HandlerFunc(api.DoActions)
+	apiRouter.ServeHTTP(rr, req)
+
+	assert.NotEqual(t, http.StatusOK, rr.Code)
+	// No new identity and no credential appended to the existing one.
+	assert.Len(t, api.mockConfig.Identities, 1)
+	assert.Len(t, api.mockConfig.Identities[0].Credentials, 0)
+}
+
+// TestEmbeddedIamCreateAccessKeyWithCallerSuppliedKeys tests creating an access key with caller-supplied credentials
+func TestEmbeddedIamCreateAccessKeyWithCallerSuppliedKeys(t *testing.T) {
+	api := NewEmbeddedIamApiForTest()
+	api.mockConfig = &iam_pb.S3ApiConfiguration{
+		Identities: []*iam_pb.Identity{
+			{Name: "TestUser"},
+		},
+	}
+
+	form := url.Values{}
+	form.Set("Action", "CreateAccessKey")
+	form.Set("UserName", "TestUser")
+	form.Set("AccessKeyId", "myapp")
+	form.Set("SecretAccessKey", "mysecret123")
+
+	req, _ := http.NewRequest("POST", "/", nil)
+	req.PostForm = form
+	req.Form = form
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	rr := httptest.NewRecorder()
+	apiRouter := mux.NewRouter().SkipClean(true)
+	apiRouter.Path("/").Methods(http.MethodPost).HandlerFunc(api.DoActions)
+	apiRouter.ServeHTTP(rr, req)
+
+	assert.Equal(t, http.StatusOK, rr.Code)
+
+	// Verify caller-supplied keys were used, not random ones
+	var out iamCreateAccessKeyResponse
+	err := xml.Unmarshal(rr.Body.Bytes(), &out)
+	require.NoError(t, err, "failed to unmarshal CreateAccessKey response")
+	require.NotNil(t, out.CreateAccessKeyResult.AccessKey.AccessKeyId)
+	require.NotNil(t, out.CreateAccessKeyResult.AccessKey.SecretAccessKey)
+	require.NotNil(t, out.CreateAccessKeyResult.AccessKey.UserName)
+	assert.Equal(t, "myapp", *out.CreateAccessKeyResult.AccessKey.AccessKeyId)
+	assert.Equal(t, "mysecret123", *out.CreateAccessKeyResult.AccessKey.SecretAccessKey)
+	assert.Equal(t, "TestUser", *out.CreateAccessKeyResult.AccessKey.UserName)
+
+	// Verify credentials were persisted with caller-supplied keys
+	require.Len(t, api.mockConfig.Identities, 1)
+	require.Len(t, api.mockConfig.Identities[0].Credentials, 1)
+	assert.Equal(t, "myapp", api.mockConfig.Identities[0].Credentials[0].AccessKey)
+	assert.Equal(t, "mysecret123", api.mockConfig.Identities[0].Credentials[0].SecretKey)
+}
+
+// TestEmbeddedIamCreateAccessKeyRejectsWeakKeys tests that weak caller-supplied keys are rejected
+func TestEmbeddedIamCreateAccessKeyRejectsWeakKeys(t *testing.T) {
+	api := NewEmbeddedIamApiForTest()
+	api.mockConfig = &iam_pb.S3ApiConfiguration{
+		Identities: []*iam_pb.Identity{
+			{Name: "TestUser"},
+		},
+	}
+
+	// AccessKeyId too short
+	form := url.Values{}
+	form.Set("Action", "CreateAccessKey")
+	form.Set("UserName", "TestUser")
+	form.Set("AccessKeyId", "ab")
+	form.Set("SecretAccessKey", "validsecret123")
+
+	req, _ := http.NewRequest("POST", "/", nil)
+	req.PostForm = form
+	req.Form = form
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	rr := httptest.NewRecorder()
+	apiRouter := mux.NewRouter().SkipClean(true)
+	apiRouter.Path("/").Methods(http.MethodPost).HandlerFunc(api.DoActions)
+	apiRouter.ServeHTTP(rr, req)
+
+	assert.NotEqual(t, http.StatusOK, rr.Code)
+	assert.Contains(t, rr.Body.String(), "AccessKeyId must be 4 to 128 alphanumeric characters")
+
+	// SecretAccessKey too short
+	form.Set("AccessKeyId", "validkey")
+	form.Set("SecretAccessKey", "short")
+
+	req, _ = http.NewRequest("POST", "/", nil)
+	req.PostForm = form
+	req.Form = form
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	rr = httptest.NewRecorder()
+	apiRouter.ServeHTTP(rr, req)
+
+	assert.NotEqual(t, http.StatusOK, rr.Code)
+	assert.Contains(t, rr.Body.String(), "SecretAccessKey must be between 8 and 128 characters")
+	// AccessKeyId with SigV4 delimiters
+	form.Set("AccessKeyId", "foo/bar=baz")
+	form.Set("SecretAccessKey", "validsecret123")
+
+	req, _ = http.NewRequest("POST", "/", nil)
+	req.PostForm = form
+	req.Form = form
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	rr = httptest.NewRecorder()
+	apiRouter.ServeHTTP(rr, req)
+
+	assert.NotEqual(t, http.StatusOK, rr.Code)
+	assert.Contains(t, rr.Body.String(), "AccessKeyId must be 4 to 128 alphanumeric characters")
+}
+
+// TestEmbeddedIamCreateAccessKeyRejectsCollision tests that duplicate access keys are rejected
+func TestEmbeddedIamCreateAccessKeyRejectsCollision(t *testing.T) {
+	api := NewEmbeddedIamApiForTest()
+	// Use a distinctive owner name ("ownerAlpha") so the leak assertion
+	// cannot accidentally match a word embedded in the error body.
+	const ownerName = "ownerAlpha"
+	api.mockConfig = &iam_pb.S3ApiConfiguration{
+		Identities: []*iam_pb.Identity{
+			{
+				Name: ownerName,
+				Credentials: []*iam_pb.Credential{
+					{AccessKey: "takenkey", SecretKey: "existingsecret"},
+				},
+			},
+			{Name: "NewUser"},
+		},
+	}
+
+	form := url.Values{}
+	form.Set("Action", "CreateAccessKey")
+	form.Set("UserName", "NewUser")
+	form.Set("AccessKeyId", "takenkey")
+	form.Set("SecretAccessKey", "newsecret123")
+
+	req, _ := http.NewRequest("POST", "/", nil)
+	req.PostForm = form
+	req.Form = form
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	rr := httptest.NewRecorder()
+	apiRouter := mux.NewRouter().SkipClean(true)
+	apiRouter.Path("/").Methods(http.MethodPost).HandlerFunc(api.DoActions)
+	apiRouter.ServeHTTP(rr, req)
+
+	assert.NotEqual(t, http.StatusOK, rr.Code)
+	assert.Contains(t, rr.Body.String(), "already in use")
+	assert.NotContains(t, rr.Body.String(), ownerName, "should not leak owner name")
+
+	// Verify no credentials were added to NewUser. Look up by name because the
+	// memory store backs LoadConfiguration with a map, so Identities order is
+	// not stable across a save/load round trip.
+	var newUser *iam_pb.Identity
+	for _, ident := range api.mockConfig.Identities {
+		if ident.Name == "NewUser" {
+			newUser = ident
+			break
+		}
+	}
+	require.NotNil(t, newUser, "NewUser identity should still exist")
+	assert.Len(t, newUser.Credentials, 0)
+}
+
+// TestEmbeddedIamCreateAccessKeyRejectsPartialSupply tests that supplying only
+// one of AccessKeyId / SecretAccessKey is rejected.
+func TestEmbeddedIamCreateAccessKeyRejectsPartialSupply(t *testing.T) {
+	api := NewEmbeddedIamApiForTest()
+	api.mockConfig = &iam_pb.S3ApiConfiguration{
+		Identities: []*iam_pb.Identity{
+			{Name: "TestUser"},
+		},
+	}
+
+	apiRouter := mux.NewRouter().SkipClean(true)
+	apiRouter.Path("/").Methods(http.MethodPost).HandlerFunc(api.DoActions)
+
+	// AccessKeyId supplied, SecretAccessKey omitted
+	form := url.Values{}
+	form.Set("Action", "CreateAccessKey")
+	form.Set("UserName", "TestUser")
+	form.Set("AccessKeyId", "myappkey")
+
+	req, _ := http.NewRequest("POST", "/", nil)
+	req.PostForm = form
+	req.Form = form
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	rr := httptest.NewRecorder()
+	apiRouter.ServeHTTP(rr, req)
+	assert.NotEqual(t, http.StatusOK, rr.Code)
+	assert.Contains(t, rr.Body.String(), "AccessKeyId and SecretAccessKey must be supplied together")
+	assert.Len(t, api.mockConfig.Identities[0].Credentials, 0)
+
+	// SecretAccessKey supplied, AccessKeyId omitted
+	form = url.Values{}
+	form.Set("Action", "CreateAccessKey")
+	form.Set("UserName", "TestUser")
+	form.Set("SecretAccessKey", "validsecret1")
+
+	req, _ = http.NewRequest("POST", "/", nil)
+	req.PostForm = form
+	req.Form = form
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	rr = httptest.NewRecorder()
+	apiRouter.ServeHTTP(rr, req)
+	assert.NotEqual(t, http.StatusOK, rr.Code)
+	assert.Contains(t, rr.Body.String(), "AccessKeyId and SecretAccessKey must be supplied together")
+	assert.Len(t, api.mockConfig.Identities[0].Credentials, 0)
+}
+
+// TestEmbeddedIamCreateAccessKeyBoundary tests key length boundaries
+func TestEmbeddedIamCreateAccessKeyBoundary(t *testing.T) {
+	api := NewEmbeddedIamApiForTest()
+	api.mockConfig = &iam_pb.S3ApiConfiguration{
+		Identities: []*iam_pb.Identity{
+			{Name: "TestUser"},
+		},
+	}
+
+	apiRouter := mux.NewRouter().SkipClean(true)
+	apiRouter.Path("/").Methods(http.MethodPost).HandlerFunc(api.DoActions)
+
+	// Exactly 4 chars — should pass
+	form := url.Values{}
+	form.Set("Action", "CreateAccessKey")
+	form.Set("UserName", "TestUser")
+	form.Set("AccessKeyId", "abcd")
+	form.Set("SecretAccessKey", "validsecret1")
+
+	req, _ := http.NewRequest("POST", "/", nil)
+	req.PostForm = form
+	req.Form = form
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	rr := httptest.NewRecorder()
+	apiRouter.ServeHTTP(rr, req)
+	assert.Equal(t, http.StatusOK, rr.Code)
+
+	// Exactly 3 chars — should fail
+	api.mockConfig.Identities[0].Credentials = nil
+	form.Set("AccessKeyId", "abc")
+
+	req, _ = http.NewRequest("POST", "/", nil)
+	req.PostForm = form
+	req.Form = form
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	rr = httptest.NewRecorder()
+	apiRouter.ServeHTTP(rr, req)
+	assert.NotEqual(t, http.StatusOK, rr.Code)
+	assert.Contains(t, rr.Body.String(), "alphanumeric")
+
+	// Exactly 128 chars — should pass
+	api.mockConfig.Identities[0].Credentials = nil
+	ak128 := strings.Repeat("a", 128)
+	sk128 := strings.Repeat("s", 128)
+	form.Set("AccessKeyId", ak128)
+	form.Set("SecretAccessKey", sk128)
+
+	req, _ = http.NewRequest("POST", "/", nil)
+	req.PostForm = form
+	req.Form = form
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	rr = httptest.NewRecorder()
+	apiRouter.ServeHTTP(rr, req)
+	assert.Equal(t, http.StatusOK, rr.Code)
+
+	// 129 chars AccessKeyId — should fail
+	api.mockConfig.Identities[0].Credentials = nil
+	form.Set("AccessKeyId", strings.Repeat("a", 129))
+	form.Set("SecretAccessKey", sk128)
+
+	req, _ = http.NewRequest("POST", "/", nil)
+	req.PostForm = form
+	req.Form = form
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	rr = httptest.NewRecorder()
+	apiRouter.ServeHTTP(rr, req)
+	assert.NotEqual(t, http.StatusOK, rr.Code)
+	assert.Contains(t, rr.Body.String(), "alphanumeric")
+
+	// 7-char SecretAccessKey — should fail
+	api.mockConfig.Identities[0].Credentials = nil
+	form.Set("AccessKeyId", "validkey")
+	form.Set("SecretAccessKey", "1234567")
+
+	req, _ = http.NewRequest("POST", "/", nil)
+	req.PostForm = form
+	req.Form = form
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	rr = httptest.NewRecorder()
+	apiRouter.ServeHTTP(rr, req)
+	assert.NotEqual(t, http.StatusOK, rr.Code)
+	assert.Contains(t, rr.Body.String(), "SecretAccessKey must be between 8 and 128 characters")
+
+	// Exactly 8-char SecretAccessKey — should pass (lower boundary)
+	api.mockConfig.Identities[0].Credentials = nil
+	form.Set("AccessKeyId", "validkey")
+	form.Set("SecretAccessKey", "12345678")
+
+	req, _ = http.NewRequest("POST", "/", nil)
+	req.PostForm = form
+	req.Form = form
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	rr = httptest.NewRecorder()
+	apiRouter.ServeHTTP(rr, req)
+	assert.Equal(t, http.StatusOK, rr.Code)
+
+	// 129-char SecretAccessKey — should fail
+	api.mockConfig.Identities[0].Credentials = nil
+	form.Set("AccessKeyId", "validkey")
+	form.Set("SecretAccessKey", strings.Repeat("s", 129))
+
+	req, _ = http.NewRequest("POST", "/", nil)
+	req.PostForm = form
+	req.Form = form
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	rr = httptest.NewRecorder()
+	apiRouter.ServeHTTP(rr, req)
+	assert.NotEqual(t, http.StatusOK, rr.Code)
+	assert.Contains(t, rr.Body.String(), "SecretAccessKey must be between 8 and 128 characters")
+}
+
 // TestEmbeddedIamDeleteAccessKey tests deleting an access key via direct form post
 func TestEmbeddedIamDeleteAccessKey(t *testing.T) {
 	api := NewEmbeddedIamApiForTest()

weed/server/filer_grpc_server_remote.go

@@ -21,30 +21,44 @@ import (
 )
 
 func (fs *FilerServer) CacheRemoteObjectToLocalCluster(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) {
-	// Use singleflight to deduplicate concurrent caching requests for the same object
+	// Use singleflight to deduplicate concurrent caching requests for the same object.
 	// This benefits all clients: S3 API, filer HTTP, Hadoop, etc.
 	cacheKey := req.Directory + "/" + req.Name
 
-	result, err, shared := fs.remoteCacheGroup.Do(cacheKey, func() (interface{}, error) {
-		return fs.doCacheRemoteObjectToLocalCluster(ctx, req)
+	// Detach from caller ctx: on failure the error path deletes every chunk
+	// already written, so cancelling mid-download loses all progress. For
+	// blobs large enough that the download outlasts the caller's timeout
+	// the retry loop never converges.
+	bgCtx := context.WithoutCancel(ctx)
+
+	// DoChan (vs Do) so the caller can bail out on ctx.Done() while the
+	// singleflight goroutine keeps caching on bgCtx; otherwise this handler
+	// goroutine stays blocked for the full download after the client is gone.
+	ch := fs.remoteCacheGroup.DoChan(cacheKey, func() (interface{}, error) {
+		return fs.doCacheRemoteObjectToLocalCluster(bgCtx, req)
 	})
 
-	if shared {
-		glog.V(2).Infof("CacheRemoteObjectToLocalCluster: shared result for %s", cacheKey)
+	select {
+	case <-ctx.Done():
+		// Caller gave up; the detached cache keeps running and a later
+		// request will find the entry cached (or join the same singleflight).
+		return nil, ctx.Err()
+	case res := <-ch:
+		if res.Shared {
+			glog.V(2).Infof("CacheRemoteObjectToLocalCluster: shared result for %s", cacheKey)
+		}
+		if res.Err != nil {
+			return nil, res.Err
+		}
+		if res.Val == nil {
+			return nil, fmt.Errorf("unexpected nil result from singleflight")
+		}
+		resp, ok := res.Val.(*filer_pb.CacheRemoteObjectToLocalClusterResponse)
+		if !ok {
+			return nil, fmt.Errorf("unexpected result type from singleflight")
+		}
+		return resp, nil
 	}
-
-	if err != nil {
-		return nil, err
-	}
-	if result == nil {
-		return nil, fmt.Errorf("unexpected nil result from singleflight")
-	}
-
-	resp, ok := result.(*filer_pb.CacheRemoteObjectToLocalClusterResponse)
-	if !ok {
-		return nil, fmt.Errorf("unexpected result type from singleflight")
-	}
-	return resp, nil
 }
 
 // doCacheRemoteObjectToLocalCluster performs the actual caching operation.

weed/server/volume_grpc_copy.go

@@ -561,6 +561,16 @@ func (vs *VolumeServer) ReceiveFile(stream volume_server_pb.VolumeServer_Receive
 				fileInfo.VolumeId, fileInfo.Ext, fileInfo.Collection, fileInfo.ShardId, fileInfo.FileSize)
 
 			if fileInfo.IsEcVolume {
+				// os.Create below truncates in place; a mounted EcVolume
+				// holds fds on the same inodes, so overwriting corrupts
+				// live readers.
+				if _, mounted := vs.store.FindEcVolume(needle.VolumeId(fileInfo.VolumeId)); mounted {
+					glog.Errorf("ReceiveFile: ec volume %d is mounted; refusing overwrite for %s", fileInfo.VolumeId, fileInfo.Ext)
+					return stream.SendAndClose(&volume_server_pb.ReceiveFileResponse{
+						Error: fmt.Sprintf("ec volume %d is mounted; unmount before ReceiveFile", fileInfo.VolumeId),
+					})
+				}
+
 				// disk_id=0 means "unset" (protobuf default), so auto-select
 				// mirrors VolumeEcShardsCopy: prefer a disk already holding
 				// this volume's shards, then any HDD, then any disk.

weed/server/volume_grpc_read_all.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 
 	"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/needle"
 )
@@ -12,6 +13,7 @@ func (vs *VolumeServer) ReadAllNeedles(req *volume_server_pb.ReadAllNeedlesReque
 
 	for _, vid := range req.VolumeIds {
 		if err := vs.streamReadOneVolume(needle.VolumeId(vid), stream); err != nil {
+			stats.VolumeServerFileReadFailures.Inc()
 			return err
 		}
 	}

weed/server/volume_grpc_read_write.go

@@ -5,20 +5,28 @@ import (
 	"fmt"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 )
 
 func (vs *VolumeServer) ReadNeedleBlob(ctx context.Context, req *volume_server_pb.ReadNeedleBlobRequest) (resp *volume_server_pb.ReadNeedleBlobResponse, err error) {
-	resp = &volume_server_pb.ReadNeedleBlobResponse{}
-	v := vs.store.GetVolume(needle.VolumeId(req.VolumeId))
-	if v == nil {
-		return nil, fmt.Errorf("not found volume id %d", req.VolumeId)
-	}
+	if err := func() error {
+		resp = &volume_server_pb.ReadNeedleBlobResponse{}
+		v := vs.store.GetVolume(needle.VolumeId(req.VolumeId))
+		if v == nil {
+			return fmt.Errorf("not found volume id %d", req.VolumeId)
+		}
 
-	resp.NeedleBlob, err = v.ReadNeedleBlob(req.Offset, types.Size(req.Size))
-	if err != nil {
-		return nil, fmt.Errorf("read needle blob offset %d size %d: %v", req.Offset, req.Size, err)
+		resp.NeedleBlob, err = v.ReadNeedleBlob(req.Offset, types.Size(req.Size))
+		if err != nil {
+			return fmt.Errorf("read needle blob offset %d size %d: %v", req.Offset, req.Size, err)
+		}
+
+		return nil
+	}(); err != nil {
+		stats.VolumeServerFileReadFailures.Inc()
+		return nil, err
 	}
 
 	return resp, nil
@@ -28,29 +36,37 @@ func (vs *VolumeServer) ReadNeedleMeta(ctx context.Context, req *volume_server_p
 	resp = &volume_server_pb.ReadNeedleMetaResponse{}
 	volumeId := needle.VolumeId(req.VolumeId)
 
-	n := &needle.Needle{
-		Id:    types.NeedleId(req.NeedleId),
-		Flags: 0x08,
-	}
-	size := req.Size
-	offset := req.Offset
+	if err := func() error {
+		n := &needle.Needle{
+			Id:    types.NeedleId(req.NeedleId),
+			Flags: 0x08,
+		}
+		size := req.Size
+		offset := req.Offset
 
-	hasVolume := vs.store.HasVolume(volumeId)
-	if !hasVolume {
-		return nil, fmt.Errorf("not found volume id %d and read needle metadata at ec shards is not supported", req.VolumeId)
-	}
-	err = vs.store.ReadVolumeNeedleMetaAt(volumeId, n, offset, size)
-	if err != nil {
+		hasVolume := vs.store.HasVolume(volumeId)
+		if !hasVolume {
+			return fmt.Errorf("not found volume id %d and read needle metadata at ec shards is not supported", req.VolumeId)
+		}
+		err = vs.store.ReadVolumeNeedleMetaAt(volumeId, n, offset, size)
+		if err != nil {
+			return err
+		}
+
+		resp.Cookie = uint32(n.Cookie)
+		resp.LastModified = n.LastModified
+		resp.Crc = n.Checksum.Value()
+		if n.HasTtl() {
+			resp.Ttl = n.Ttl.String()
+		}
+		resp.AppendAtNs = n.AppendAtNs
+
+		return nil
+	}(); err != nil {
+		stats.VolumeServerFileReadFailures.Inc()
 		return nil, err
 	}
 
-	resp.Cookie = uint32(n.Cookie)
-	resp.LastModified = n.LastModified
-	resp.Crc = n.Checksum.Value()
-	if n.HasTtl() {
-		resp.Ttl = n.Ttl.String()
-	}
-	resp.AppendAtNs = n.AppendAtNs
 	return resp, nil
 }
 
@@ -61,13 +77,19 @@ func (vs *VolumeServer) WriteNeedleBlob(ctx context.Context, req *volume_server_
 
 	resp = &volume_server_pb.WriteNeedleBlobResponse{}
 
-	v := vs.store.GetVolume(needle.VolumeId(req.VolumeId))
-	if v == nil {
-		return nil, fmt.Errorf("not found volume id %d", req.VolumeId)
-	}
+	if err := func() error {
+		v := vs.store.GetVolume(needle.VolumeId(req.VolumeId))
+		if v == nil {
+			return fmt.Errorf("not found volume id %d", req.VolumeId)
+		}
+		if err = v.WriteNeedleBlob(types.NeedleId(req.NeedleId), req.NeedleBlob, types.Size(req.Size)); err != nil {
+			return fmt.Errorf("write blob needle %d size %d: %v", req.NeedleId, req.Size, err)
+		}
 
-	if err = v.WriteNeedleBlob(types.NeedleId(req.NeedleId), req.NeedleBlob, types.Size(req.Size)); err != nil {
-		return nil, fmt.Errorf("write blob needle %d size %d: %v", req.NeedleId, req.Size, err)
+		return nil
+	}(); err != nil {
+		stats.VolumeServerFileWriteFailures.Inc()
+		return nil, err
 	}
 
 	return resp, nil

weed/server/volume_server_handlers_write.go

@@ -57,6 +57,7 @@ func (vs *VolumeServer) PostHandler(w http.ResponseWriter, r *http.Request) {
 	// http 204 status code does not allow body
 	if writeError == nil && isUnchanged {
 		SetEtag(w, reqNeedle.Etag())
+		w.Header().Set("Content-MD5", contentMd5)
 		w.WriteHeader(http.StatusNoContent)
 		return
 	}

weed/stats/metrics.go

@@ -370,6 +370,22 @@ var (
 			Help:      "Number of master server disconnections.",
 		}, []string{"address"})
 
+	VolumeServerFileReadFailures = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: Namespace,
+			Subsystem: "volumeServer",
+			Name:      "file_read_failures",
+			Help:      "Counter of overall failed file read requests from clients.",
+		})
+
+	VolumeServerFileWriteFailures = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: Namespace,
+			Subsystem: "volumeServer",
+			Name:      "file_write_failures",
+			Help:      "Counter of overall failed file write requests from clients.",
+		})
+
 	S3RequestCounter = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: Namespace,
@@ -537,6 +553,8 @@ func init() {
 	Gather.MustRegister(VolumeServerInFlightDownloadSize)
 	Gather.MustRegister(VolumeServerInFlightUploadSize)
 	Gather.MustRegister(VolumeServerMasterDisconnections)
+	Gather.MustRegister(VolumeServerFileReadFailures)
+	Gather.MustRegister(VolumeServerFileWriteFailures)
 
 	Gather.MustRegister(S3RequestCounter)
 	Gather.MustRegister(S3HandlerCounter)

weed/worker/tasks/erasure_coding/ec_task.go

@@ -596,7 +596,6 @@ func (t *ErasureCodingTask) deleteOriginalVolume(ctx context.Context) error {
 		}
 	}
 
-	// Report results
 	if len(deleteErrors) > 0 {
 		t.GetLogger().WithFields(map[string]interface{}{
 			"volume_id":      t.volumeID,
@@ -605,16 +604,19 @@ func (t *ErasureCodingTask) deleteOriginalVolume(ctx context.Context) error {
 			"total_replicas": len(replicas),
 			"success_rate":   float64(successCount) / float64(len(replicas)) * 100,
 			"errors":         deleteErrors,
-		}).Warning("Some volume deletions failed")
-		// Don't return error - EC task should still be considered successful if shards are mounted
-	} else {
-		t.GetLogger().WithFields(map[string]interface{}{
-			"volume_id":       t.volumeID,
-			"replica_count":   len(replicas),
-			"replica_servers": replicas,
-		}).Info("Successfully deleted volume from all replica servers")
+		}).Error("Failed to delete some original volume replicas after EC encoding")
+		// A surviving source replica lets a later detection scan re-propose
+		// EC on the same volume, which retries over mounted shards.
+		return fmt.Errorf("failed to delete %d of %d original volume replicas for volume %d: %s",
+			len(deleteErrors), len(replicas), t.volumeID, strings.Join(deleteErrors, "; "))
 	}
 
+	t.GetLogger().WithFields(map[string]interface{}{
+		"volume_id":       t.volumeID,
+		"replica_count":   len(replicas),
+		"replica_servers": replicas,
+	}).Info("Successfully deleted volume from all replica servers")
+
 	return nil
 }
 

weed/worker/tasks/erasure_coding/ec_task_delete_swallow_test.go

@@ -0,0 +1,113 @@
+package erasure_coding
+
+import (
+	"context"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/test/volume_server/framework"
+	"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+// One reachable replica + one unreachable: the reachable delete still
+// succeeds, and the function surfaces an error naming the failure.
+func TestDeleteOriginalVolumeSurfacesReplicaFailures(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	clusterHarness := framework.StartVolumeCluster(t, matrix.P1())
+	conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress())
+	defer conn.Close()
+
+	const volumeID = uint32(91842)
+	framework.AllocateVolume(t, grpcClient, volumeID, "")
+
+	httpClient := framework.NewHTTPClient()
+	fid := framework.NewFileID(volumeID, 918420, 0x91842042)
+	uploadResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid,
+		[]byte("delete-surface-content-for-issue-9184"))
+	_ = framework.ReadAllAndClose(t, uploadResp)
+	require.Equal(t, http.StatusCreated, uploadResp.StatusCode)
+
+	task := NewErasureCodingTask(
+		"delete-surface-fix",
+		clusterHarness.VolumeServerAddress(),
+		volumeID,
+		"",
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+
+	unreachable := "127.0.0.1:1"
+	task.sources = []*worker_pb.TaskSource{
+		{
+			Node:     clusterHarness.VolumeServerAddress(),
+			VolumeId: volumeID,
+		},
+		{
+			Node:     unreachable,
+			VolumeId: volumeID,
+		},
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	err := task.deleteOriginalVolume(ctx)
+	require.Error(t, err, "deleteOriginalVolume must surface replica delete failures (#9184)")
+	require.Contains(t, err.Error(), unreachable,
+		"returned error should name the replica that failed: %v", err)
+	require.True(t,
+		strings.Contains(err.Error(), "failed to delete"),
+		"returned error should describe what failed: %v", err)
+
+	_, statusErr := grpcClient.VolumeStatus(ctx, &volume_server_pb.VolumeStatusRequest{VolumeId: volumeID})
+	require.Error(t, statusErr,
+		"reachable replica %d should have been deleted before failure was surfaced", volumeID)
+}
+
+func TestDeleteOriginalVolumeSucceedsWhenAllReplicasReachable(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	clusterHarness := framework.StartVolumeCluster(t, matrix.P1())
+	conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress())
+	defer conn.Close()
+
+	const volumeID = uint32(91844)
+	framework.AllocateVolume(t, grpcClient, volumeID, "")
+
+	httpClient := framework.NewHTTPClient()
+	fid := framework.NewFileID(volumeID, 918440, 0x91844042)
+	uploadResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid,
+		[]byte("delete-happy-path-content-for-issue-9184"))
+	_ = framework.ReadAllAndClose(t, uploadResp)
+	require.Equal(t, http.StatusCreated, uploadResp.StatusCode)
+
+	task := NewErasureCodingTask(
+		"delete-happy-path",
+		clusterHarness.VolumeServerAddress(),
+		volumeID,
+		"",
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	task.sources = []*worker_pb.TaskSource{
+		{Node: clusterHarness.VolumeServerAddress(), VolumeId: volumeID},
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	require.NoError(t, task.deleteOriginalVolume(ctx))
+
+	_, statusErr := grpcClient.VolumeStatus(ctx, &volume_server_pb.VolumeStatusRequest{VolumeId: volumeID})
+	require.Error(t, statusErr, "volume %d should be gone after successful delete", volumeID)
+}