4.19

* filer: bootstrap pre-existing metadata when a new filer joins a cluster

When a filer connects to a peer for the first time (no stored sync
offset), it now does a full BFS traversal of the peer's metadata via
TraverseBfsMetadata before starting the incremental change stream.
This ensures filer2 sees all data that existed before it started,
fixing the issue where only post-startup changes were synced.

Closes #8961

* filer: upsert during bootstrap and persist offset immediately

- Use upsert (insert, then update on conflict) during metadata
  traversal so the bootstrap doesn't fail on the root directory
  or after a partial previous attempt.
- Persist the sync offset right after a successful traversal so
  a retry doesn't redo the full BFS.

* filer: address review feedback on metadata bootstrap

- Use peer-side max Mtime as the streaming cursor instead of local
  time.Now() to avoid missing events due to clock skew between filers.
  traversePeerMetadata now returns the high-water Mtime (nanoseconds)
  observed during BFS traversal.

- Compare Mtime before overwriting during bootstrap: if a local entry
  is newer than the peer's version, skip the update instead of
  clobbering it.

- Only trigger full BFS traversal on ErrKvNotFound (key genuinely
  missing). Transient KvGet errors (connection issues, etc.) are now
  propagated instead of silently falling through to a full re-sync.
  Changed readOffset to use %w so errors.Is works through the chain.

* filer: address review findings on bootstrap sync

- Use wall-clock time with safety margin for stream cursor instead of
  entry Mtime. Mtime is file modification time (can be arbitrary),
  while the metadata stream uses TsNs (event log time). Using
  time.Now() minus 1 minute before traversal ensures no events are
  missed even with clock skew, matching the proven filer.meta.backup
  pattern.

- Pass ExcludedPrefixes=[SystemLogDir] to TraverseBfsMetadata so
  the server prunes internal log entries server-side instead of
  transferring them over the network only to be filtered client-side.

- Fail fast if updateOffset fails after bootstrap. If we can't
  persist the offset, bail out rather than proceeding and potentially
  losing the expensive BFS work on the next retry.

k8s/charts/seaweedfs/Chart.yaml

@@ -1,6 +1,6 @@
 apiVersion: v1
 description: SeaweedFS
 name: seaweedfs
-appVersion: "4.18"
+appVersion: "4.19"
 # Dev note: Trigger a helm chart release by `git tag -a helm-<version>`
-version: 4.18.0
+version: 4.19.0

test/plugin_workers/volume_fixtures.go

@@ -7,6 +7,7 @@ import (
 	"path/filepath"
 	"testing"
 
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 )
 
@@ -21,7 +22,12 @@ func WriteTestVolumeFiles(t *testing.T, baseDir string, volumeID uint32, datSize
 	datPath := filepath.Join(baseDir, volumeFilename(volumeID, ".dat"))
 	idxPath := filepath.Join(baseDir, volumeFilename(volumeID, ".idx"))
 
-	data := make([]byte, datSize)
+	// The idx entry records the needle's data size. The actual on-disk size
+	// includes header + checksum + timestamp (GetActualSize). The .dat must
+	// be large enough to hold the full needle.
+	needleDataSize := types.Size(datSize)
+	actualSize := needle.GetActualSize(needleDataSize, needle.Version3)
+	data := make([]byte, actualSize)
 	rng := rand.New(rand.NewSource(99))
 	_, _ = rng.Read(data)
 	if err := os.WriteFile(datPath, data, 0644); err != nil {
@@ -35,7 +41,7 @@ func WriteTestVolumeFiles(t *testing.T, baseDir string, volumeID uint32, datSize
 
 	types.NeedleIdToBytes(entry[:idEnd], types.NeedleId(1))
 	types.OffsetToBytes(entry[idEnd:offsetEnd], types.ToOffset(0))
-	types.SizeToBytes(entry[offsetEnd:sizeEnd], types.Size(datSize))
+	types.SizeToBytes(entry[offsetEnd:sizeEnd], needleDataSize)
 
 	if err := os.WriteFile(idxPath, entry, 0644); err != nil {
 		t.Fatalf("write idx file: %v", err)

test/s3/spark/issue_8285_repro_test.go

@@ -70,7 +70,8 @@ print("WRITE_COUNT=" + str(count))
 		"issue-8285/output/_temporary/0/",
 		"issue-8285/output/_temporary/0/_temporary/",
 	}
-	lingering := waitForObjectsToDisappear(t, env, "test", temporaryCandidates, 35*time.Second)
+	// Empty folder cleanup has a 2m default delay + 30s processor interval
+	lingering := waitForObjectsToDisappear(t, env, "test", temporaryCandidates, 3*time.Minute)
 	if len(lingering) > 0 {
 		t.Fatalf("issue #8285 regression detected: lingering temporary directories: %v", lingering)
 	}

test/s3tables/catalog_spark/spark_operations_test.go

@@ -277,3 +277,80 @@ print(f"Count at snapshot: {count}")
 
 	t.Logf(">>> Time travel test passed")
 }
+
+// TestSparkMultiLevelNamespace tests that multi-level namespaces produce correct
+// S3 paths (dot-separated) so that Spark can read back the data it writes.
+// Regression test for https://github.com/seaweedfs/seaweedfs/issues/8959
+func TestSparkMultiLevelNamespace(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	env, _, _ := setupSparkTestEnv(t)
+
+	// Use a two-level namespace like "analytics.daily"
+	nsLevel1 := "analytics_" + randomString(4)
+	nsLevel2 := "daily_" + randomString(4)
+	multiNs := fmt.Sprintf("%s.%s", nsLevel1, nsLevel2)
+	tableName := "events_" + randomString(4)
+
+	// Create multi-level namespace
+	t.Logf(">>> Creating multi-level namespace: %s", multiNs)
+	createNsSQL := fmt.Sprintf(`
+spark.sql("CREATE NAMESPACE iceberg.%s")
+print("Namespace created")
+`, multiNs)
+	output := runSparkPySQL(t, env.sparkContainer, createNsSQL, env.icebergRestPort, env.s3Port)
+	if !strings.Contains(output, "Namespace created") {
+		t.Fatalf("multi-level namespace creation failed, output: %s", output)
+	}
+
+	// Create table under multi-level namespace
+	t.Logf(">>> Creating table under multi-level namespace")
+	createTableSQL := fmt.Sprintf(`
+spark.sql("""
+CREATE TABLE iceberg.%s.%s (
+    id INT,
+    event STRING,
+    ts TIMESTAMP
+)
+USING iceberg
+""")
+print("Table created")
+`, multiNs, tableName)
+	output = runSparkPySQL(t, env.sparkContainer, createTableSQL, env.icebergRestPort, env.s3Port)
+	if !strings.Contains(output, "Table created") {
+		t.Fatalf("table creation under multi-level namespace failed, output: %s", output)
+	}
+
+	// Insert data
+	t.Logf(">>> Inserting data into multi-level namespace table")
+	insertSQL := fmt.Sprintf(`
+spark.sql("""
+INSERT INTO iceberg.%s.%s VALUES
+    (1, 'click', TIMESTAMP '2025-01-01 00:00:00'),
+    (2, 'view',  TIMESTAMP '2025-01-01 01:00:00'),
+    (3, 'click', TIMESTAMP '2025-01-02 00:00:00')
+""")
+print("Data inserted")
+`, multiNs, tableName)
+	output = runSparkPySQL(t, env.sparkContainer, insertSQL, env.icebergRestPort, env.s3Port)
+	if !strings.Contains(output, "Data inserted") {
+		t.Fatalf("data insertion failed, output: %s", output)
+	}
+
+	// Query data back — this is the key test: if the namespace path separator
+	// was wrong (\x1F instead of "."), Spark would not find the data files.
+	t.Logf(">>> Querying data from multi-level namespace table")
+	querySQL := fmt.Sprintf(`
+result = spark.sql("SELECT COUNT(*) as count FROM iceberg.%s.%s")
+count = result.collect()[0]['count']
+print(f"Row count: {count}")
+`, multiNs, tableName)
+	output = runSparkPySQL(t, env.sparkContainer, querySQL, env.icebergRestPort, env.s3Port)
+	if !strings.Contains(output, "Row count: 3") {
+		t.Errorf("expected row count 3 from multi-level namespace table, got output: %s", output)
+	}
+
+	t.Logf(">>> Multi-level namespace test passed")
+}

test/s3tables/catalog_trino/trino_catalog_test.go

@@ -82,6 +82,101 @@ func TestTrinoIcebergCatalog(t *testing.T) {
 	runTrinoSQL(t, env.trinoContainer, fmt.Sprintf("SHOW TABLES FROM iceberg.%s", schemaName))
 }
 
+// TestTrinoMultiLevelNamespace tests that multi-level namespaces (dot-separated)
+// produce correct S3 paths so Trino can read back data it writes.
+// Regression test for https://github.com/seaweedfs/seaweedfs/issues/8959
+func TestTrinoMultiLevelNamespace(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	env := NewTestEnvironment(t)
+	defer env.Cleanup(t)
+
+	if !env.dockerAvailable {
+		t.Skip("Docker not available, skipping Trino integration test")
+	}
+
+	t.Logf(">>> Starting SeaweedFS...")
+	env.StartSeaweedFS(t)
+
+	tableBucket := "iceberg-tables"
+	createTableBucket(t, env, tableBucket)
+
+	configDir := env.writeTrinoConfig(t, tableBucket, withNestedNamespace())
+	env.startTrinoContainer(t, configDir)
+	waitForTrino(t, env.trinoContainer, 60*time.Second)
+
+	// Use a two-level namespace: "analytics.daily"
+	nsLevel1 := "analytics_" + randomString(4)
+	nsLevel2 := "daily_" + randomString(4)
+	flatNs := fmt.Sprintf("%s.%s", nsLevel1, nsLevel2)
+	// Trino uses double-quoted schema names for multi-level namespaces
+	multiNs := fmt.Sprintf(`"%s"`, flatNs)
+	tableName := "events_" + randomString(4)
+
+	// Create multi-level namespace (schema)
+	t.Logf(">>> Creating multi-level schema: %s", flatNs)
+	runTrinoSQL(t, env.trinoContainer, fmt.Sprintf("CREATE SCHEMA IF NOT EXISTS iceberg.%s", multiNs))
+
+	// Verify the schema shows up
+	output := runTrinoSQL(t, env.trinoContainer, "SHOW SCHEMAS FROM iceberg")
+	if !strings.Contains(output, flatNs) {
+		t.Fatalf("Expected schema %s in output:\n%s", flatNs, output)
+	}
+
+	// Create table with explicit location to avoid non-empty location conflict.
+	// The location uses the dot-separated namespace — if #8959 regresses
+	// (unit separator instead of dot), data would be written to the wrong path.
+	tableLocation := fmt.Sprintf("s3://%s/%s/%s_%s", tableBucket, flatNs, tableName, randomString(6))
+	t.Logf(">>> Creating table at location: %s", tableLocation)
+	createSQL := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS iceberg.%s.%s (
+		id INTEGER,
+		event VARCHAR,
+		ts TIMESTAMP(6)
+	) WITH (
+		format = 'PARQUET',
+		location = '%s'
+	)`, multiNs, tableName, tableLocation)
+	runTrinoSQLAllowExists(t, env.trinoContainer, createSQL)
+
+	// Insert data
+	t.Logf(">>> Inserting data into multi-level namespace table")
+	runTrinoSQL(t, env.trinoContainer, fmt.Sprintf(`
+		INSERT INTO iceberg.%s.%s VALUES
+			(1, 'click', TIMESTAMP '2025-01-01 00:00:00'),
+			(2, 'view',  TIMESTAMP '2025-01-01 01:00:00'),
+			(3, 'click', TIMESTAMP '2025-01-02 00:00:00')
+	`, multiNs, tableName))
+
+	// Query data back — if the namespace path separator were wrong (\x1F
+	// instead of "."), the metadata location would point to a non-existent
+	// S3 path and this query would fail.
+	t.Logf(">>> Querying data from multi-level namespace table")
+	countOutput := runTrinoSQL(t, env.trinoContainer, fmt.Sprintf(
+		"SELECT count(*) FROM iceberg.%s.%s", multiNs, tableName))
+	rowCount := mustParseCSVInt64(t, countOutput)
+	if rowCount != 3 {
+		t.Fatalf("expected row count 3, got %d", rowCount)
+	}
+
+	// Verify the S3 file path contains the dot-separated namespace, not \x1F.
+	filesOutput := runTrinoSQL(t, env.trinoContainer, fmt.Sprintf(
+		`SELECT file_path FROM iceberg.%s."%s$files" LIMIT 1`, multiNs, tableName))
+	filePath := strings.TrimSpace(filesOutput)
+	if filePath == "" {
+		t.Fatalf("expected at least one data file, got empty output")
+	}
+	if !strings.Contains(filePath, flatNs+"/") {
+		t.Errorf("expected file path to contain dot-separated namespace %q, got: %s", flatNs, filePath)
+	}
+	if strings.Contains(filePath, "\x1F") {
+		t.Errorf("file path contains unit separator (\\x1F), expected dot separator: %s", filePath)
+	}
+
+	t.Logf(">>> Trino multi-level namespace test passed")
+}
+
 func NewTestEnvironment(t *testing.T) *TestEnvironment {
 	t.Helper()
 
@@ -332,18 +427,32 @@ func testIcebergRestAPI(t *testing.T, env *TestEnvironment) {
 	}
 }
 
-func (env *TestEnvironment) writeTrinoConfig(t *testing.T, warehouseBucket string) string {
+func (env *TestEnvironment) writeTrinoConfig(t *testing.T, warehouseBucket string, opts ...func(*trinoConfigOptions)) string {
 	t.Helper()
 
-	configDir := filepath.Join(env.dataDir, "trino")
+	o := trinoConfigOptions{}
+	for _, fn := range opts {
+		fn(&o)
+	}
+
+	dirName := "trino"
+	if o.nestedNamespace {
+		dirName = "trino-nested"
+	}
+	configDir := filepath.Join(env.dataDir, dirName)
 	if err := os.MkdirAll(configDir, 0755); err != nil {
 		t.Fatalf("Failed to create Trino config dir: %v", err)
 	}
 
+	nestedLine := ""
+	if o.nestedNamespace {
+		nestedLine = "\niceberg.rest-catalog.nested-namespace-enabled=true"
+	}
+
 	config := fmt.Sprintf(`connector.name=iceberg
 iceberg.catalog.type=rest
 iceberg.rest-catalog.uri=http://host.docker.internal:%d
-iceberg.rest-catalog.warehouse=s3://%s
+iceberg.rest-catalog.warehouse=s3://%s%s
 iceberg.file-format=PARQUET
 iceberg.unique-table-location=true
 
@@ -358,7 +467,7 @@ s3.region=us-west-2
 
 # REST catalog authentication
 iceberg.rest-catalog.security=SIGV4
-`, env.icebergPort, warehouseBucket, env.s3Port, env.accessKey, env.secretKey)
+`, env.icebergPort, warehouseBucket, nestedLine, env.s3Port, env.accessKey, env.secretKey)
 
 	if err := os.WriteFile(filepath.Join(configDir, "iceberg.properties"), []byte(config), 0644); err != nil {
 		t.Fatalf("Failed to write Trino config: %v", err)
@@ -367,6 +476,14 @@ iceberg.rest-catalog.security=SIGV4
 	return configDir
 }
 
+type trinoConfigOptions struct {
+	nestedNamespace bool
+}
+
+func withNestedNamespace() func(*trinoConfigOptions) {
+	return func(o *trinoConfigOptions) { o.nestedNamespace = true }
+}
+
 func (env *TestEnvironment) startTrinoContainer(t *testing.T, configDir string) {
 	t.Helper()
 

test/volume_server/framework/volume_fixture.go

@@ -80,6 +80,42 @@ func EnableMaintenanceMode(t testing.TB, ctx context.Context, client volume_serv
 	}
 }
 
+// CorruptDatFile truncates a volume's .dat file to just the superblock (8 bytes)
+// so that needle reads fail during a full scrub due to data file size mismatch.
+func CorruptDatFile(t testing.TB, baseDir string, volumeID uint32) {
+	t.Helper()
+	datPath := filepath.Join(baseDir, "volume", fmt.Sprintf("%d.dat", volumeID))
+	// Truncate to superblock size only, removing all needle data.
+	if err := os.Truncate(datPath, 8); err != nil {
+		t.Fatalf("truncate dat file for corruption: %v", err)
+	}
+}
+
+// CorruptEcxFile appends garbage bytes to a volume's .ecx file on disk so
+// that CheckIndexFile detects a size mismatch during EC index scrub.
+func CorruptEcxFile(t testing.TB, baseDir string, volumeID uint32) {
+	t.Helper()
+	ecxPath := filepath.Join(baseDir, "volume", fmt.Sprintf("%d.ecx", volumeID))
+	f, err := os.OpenFile(ecxPath, os.O_WRONLY|os.O_APPEND, 0644)
+	if err != nil {
+		t.Fatalf("open ecx file for corruption: %v", err)
+	}
+	defer f.Close()
+	if _, err := f.Write([]byte{0xDE, 0xAD}); err != nil {
+		t.Fatalf("corrupt ecx file: %v", err)
+	}
+}
+
+// CorruptEcShardFile truncates an EC shard file to 1 byte so that local shard
+// reads fail during an EC scrub.
+func CorruptEcShardFile(t testing.TB, baseDir string, volumeID uint32, shardID int) {
+	t.Helper()
+	shardPath := filepath.Join(baseDir, "volume", fmt.Sprintf("%d.ec%02d", volumeID, shardID))
+	if err := os.Truncate(shardPath, 1); err != nil {
+		t.Fatalf("truncate EC shard file %s: %v", shardPath, err)
+	}
+}
+
 func ReadBytes(t testing.TB, client *http.Client, volumeURL, fid string) *http.Response {
 	t.Helper()
 

test/volume_server/grpc/erasure_coding_test.go

@@ -752,6 +752,87 @@ func TestEcShardsCopyFromPeerSuccess(t *testing.T) {
 	}
 }
 
+// TestEcIndexConsistencyAfterEncode verifies that every needle indexed in .ecx
+// can be read back correctly from EC shards after VolumeEcShardsGenerate.
+// This catches the race condition fixed in this PR where .ecx could reference
+// data not present in EC shards.
+func TestEcIndexConsistencyAfterEncode(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(130)
+	framework.AllocateVolume(t, grpcClient, volumeID, "")
+
+	httpClient := framework.NewHTTPClient()
+
+	// Upload multiple needles of varying sizes
+	type testNeedle struct {
+		fid     string
+		payload []byte
+	}
+	needles := []testNeedle{
+		{framework.NewFileID(volumeID, 1001, 0xAABB0001), []byte("small-needle-1")},
+		{framework.NewFileID(volumeID, 1002, 0xAABB0002), make([]byte, 1024)},     // 1KB
+		{framework.NewFileID(volumeID, 1003, 0xAABB0003), make([]byte, 64*1024)},   // 64KB
+		{framework.NewFileID(volumeID, 1004, 0xAABB0004), make([]byte, 256*1024)},  // 256KB
+		{framework.NewFileID(volumeID, 1005, 0xAABB0005), []byte("small-needle-2")},
+	}
+
+	// Fill larger payloads with recognizable data
+	for i := range needles {
+		for j := range needles[i].payload {
+			needles[i].payload[j] = byte(i*37 + j%251)
+		}
+	}
+
+	for _, n := range needles {
+		resp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), n.fid, n.payload)
+		_ = framework.ReadAllAndClose(t, resp)
+		if resp.StatusCode != http.StatusCreated {
+			t.Fatalf("upload %s expected 201, got %d", n.fid, resp.StatusCode)
+		}
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	// EC encode
+	_, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{
+		VolumeId:   volumeID,
+		Collection: "",
+	})
+	if err != nil {
+		t.Fatalf("VolumeEcShardsGenerate failed: %v", err)
+	}
+
+	// Mount all data shards so reads go through the EC path
+	_, err = grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{
+		VolumeId:   volumeID,
+		Collection: "",
+		ShardIds:   []uint32{0, 1, 2, 3, 4, 5, 6, 7, 8, 9},
+	})
+	if err != nil {
+		t.Fatalf("VolumeEcShardsMount failed: %v", err)
+	}
+
+	// Read every needle back from EC shards and verify payload
+	for _, n := range needles {
+		readResp := framework.ReadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), n.fid)
+		readBody := framework.ReadAllAndClose(t, readResp)
+		if readResp.StatusCode != http.StatusOK {
+			t.Fatalf("EC read %s expected 200, got %d", n.fid, readResp.StatusCode)
+		}
+		if string(readBody) != string(n.payload) {
+			t.Fatalf("EC read %s payload mismatch: got %d bytes, want %d bytes", n.fid, len(readBody), len(n.payload))
+		}
+	}
+}
+
 func TestEcShardsCopyFailsWhenSourceUnavailable(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping integration test in short mode")

test/volume_server/grpc/scrub_integration_test.go

@@ -0,0 +1,416 @@
+package volume_server_grpc_test
+
+import (
+	"context"
+	"net/http"
+	"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/storage/erasure_coding"
+)
+
+// --- Normal volume scrub tests ---
+
+func TestScrubVolumeFullHealthy(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	clusterHarness := framework.StartSingleVolumeCluster(t, matrix.P1())
+	conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress())
+	defer conn.Close()
+
+	const volumeID = uint32(200)
+	framework.AllocateVolume(t, grpcClient, volumeID, "")
+
+	httpClient := framework.NewHTTPClient()
+	framework.ReadAllAndClose(t, framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), framework.NewFileID(volumeID, 1, 1), []byte("data-one")))
+	framework.ReadAllAndClose(t, framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), framework.NewFileID(volumeID, 2, 2), []byte("data-two")))
+	framework.ReadAllAndClose(t, framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), framework.NewFileID(volumeID, 3, 3), []byte("data-three")))
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	resp, err := grpcClient.ScrubVolume(ctx, &volume_server_pb.ScrubVolumeRequest{
+		VolumeIds: []uint32{volumeID},
+		Mode:      volume_server_pb.VolumeScrubMode_FULL,
+	})
+	if err != nil {
+		t.Fatalf("ScrubVolume FULL on healthy volume failed: %v", err)
+	}
+	if resp.GetTotalVolumes() != 1 {
+		t.Fatalf("expected total_volumes=1, got %d", resp.GetTotalVolumes())
+	}
+	if resp.GetTotalFiles() != 3 {
+		t.Fatalf("expected total_files=3, got %d", resp.GetTotalFiles())
+	}
+	if len(resp.GetBrokenVolumeIds()) != 0 {
+		t.Fatalf("expected no broken volumes, got %v: %v", resp.GetBrokenVolumeIds(), resp.GetDetails())
+	}
+}
+
+func TestScrubVolumeFullCorruptData(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	clusterHarness := framework.StartSingleVolumeCluster(t, matrix.P1())
+	conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress())
+	defer conn.Close()
+
+	const volumeID = uint32(201)
+	framework.AllocateVolume(t, grpcClient, volumeID, "")
+
+	httpClient := framework.NewHTTPClient()
+	framework.ReadAllAndClose(t, framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), framework.NewFileID(volumeID, 1, 1), []byte("important data")))
+
+	framework.CorruptDatFile(t, clusterHarness.BaseDir(), volumeID)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	resp, err := grpcClient.ScrubVolume(ctx, &volume_server_pb.ScrubVolumeRequest{
+		VolumeIds: []uint32{volumeID},
+		Mode:      volume_server_pb.VolumeScrubMode_FULL,
+	})
+	if err != nil {
+		t.Fatalf("ScrubVolume FULL on corrupt volume failed: %v", err)
+	}
+	if len(resp.GetBrokenVolumeIds()) == 0 {
+		t.Fatalf("expected broken volume after data corruption, got none")
+	}
+	if len(resp.GetDetails()) == 0 {
+		t.Fatalf("expected error details for corrupt volume")
+	}
+}
+
+func TestScrubVolumeMixedHealthy(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	clusterHarness := framework.StartSingleVolumeCluster(t, matrix.P1())
+	conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress())
+	defer conn.Close()
+
+	const healthyVol = uint32(202)
+	const corruptVol = uint32(203)
+	framework.AllocateVolume(t, grpcClient, healthyVol, "")
+	framework.AllocateVolume(t, grpcClient, corruptVol, "")
+
+	httpClient := framework.NewHTTPClient()
+	framework.ReadAllAndClose(t, framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), framework.NewFileID(healthyVol, 1, 1), []byte("healthy")))
+	framework.ReadAllAndClose(t, framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), framework.NewFileID(corruptVol, 1, 1), []byte("will corrupt")))
+
+	framework.CorruptIndexFile(t, clusterHarness.BaseDir(), corruptVol)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	resp, err := grpcClient.ScrubVolume(ctx, &volume_server_pb.ScrubVolumeRequest{
+		VolumeIds: []uint32{healthyVol, corruptVol},
+		Mode:      volume_server_pb.VolumeScrubMode_INDEX,
+	})
+	if err != nil {
+		t.Fatalf("ScrubVolume INDEX on mixed volumes failed: %v", err)
+	}
+	if resp.GetTotalVolumes() != 2 {
+		t.Fatalf("expected total_volumes=2, got %d", resp.GetTotalVolumes())
+	}
+	if len(resp.GetBrokenVolumeIds()) != 1 {
+		t.Fatalf("expected exactly 1 broken volume, got %v", resp.GetBrokenVolumeIds())
+	}
+	if resp.GetBrokenVolumeIds()[0] != corruptVol {
+		t.Fatalf("expected broken volume %d, got %d", corruptVol, resp.GetBrokenVolumeIds()[0])
+	}
+}
+
+func TestScrubVolumeMissingVolumeReturnsError(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	clusterHarness := framework.StartSingleVolumeCluster(t, matrix.P1())
+	conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress())
+	defer conn.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	_, err := grpcClient.ScrubVolume(ctx, &volume_server_pb.ScrubVolumeRequest{
+		VolumeIds: []uint32{99999},
+		Mode:      volume_server_pb.VolumeScrubMode_FULL,
+	})
+	if err == nil {
+		t.Fatalf("ScrubVolume should fail for missing volume")
+	}
+}
+
+// --- EC volume scrub tests ---
+
+// ecSetup creates a volume, uploads data, generates EC shards, and mounts all of them.
+func ecSetup(t *testing.T, grpcClient volume_server_pb.VolumeServerClient, httpClient *http.Client, volumeURL string, volumeID uint32) {
+	t.Helper()
+	framework.AllocateVolume(t, grpcClient, volumeID, "")
+
+	fid := framework.NewFileID(volumeID, 1, 0xABCD0001)
+	uploadResp := framework.UploadBytes(t, httpClient, volumeURL, fid, []byte("ec-scrub-test-data-payload"))
+	_ = 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()
+
+	_, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{
+		VolumeId:   volumeID,
+		Collection: "",
+	})
+	if err != nil {
+		t.Fatalf("VolumeEcShardsGenerate failed: %v", err)
+	}
+
+	allShards := make([]uint32, erasure_coding.TotalShardsCount)
+	for i := range allShards {
+		allShards[i] = uint32(i)
+	}
+	_, err = grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{
+		VolumeId:   volumeID,
+		Collection: "",
+		ShardIds:   allShards,
+	})
+	if err != nil {
+		t.Fatalf("VolumeEcShardsMount all shards failed: %v", err)
+	}
+}
+
+func TestScrubEcVolumeIndexHealthy(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(210)
+	httpClient := framework.NewHTTPClient()
+	ecSetup(t, grpcClient, httpClient, clusterHarness.VolumeAdminURL(), volumeID)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	resp, err := grpcClient.ScrubEcVolume(ctx, &volume_server_pb.ScrubEcVolumeRequest{
+		VolumeIds: []uint32{volumeID},
+		Mode:      volume_server_pb.VolumeScrubMode_INDEX,
+	})
+	if err != nil {
+		t.Fatalf("ScrubEcVolume INDEX on healthy volume failed: %v", err)
+	}
+	if resp.GetTotalVolumes() != 1 {
+		t.Fatalf("expected total_volumes=1, got %d", resp.GetTotalVolumes())
+	}
+	if len(resp.GetBrokenVolumeIds()) != 0 {
+		t.Fatalf("expected no broken volumes, got %v: %v", resp.GetBrokenVolumeIds(), resp.GetDetails())
+	}
+}
+
+func TestScrubEcVolumeLocalHealthy(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(211)
+	httpClient := framework.NewHTTPClient()
+	ecSetup(t, grpcClient, httpClient, clusterHarness.VolumeAdminURL(), volumeID)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	resp, err := grpcClient.ScrubEcVolume(ctx, &volume_server_pb.ScrubEcVolumeRequest{
+		VolumeIds: []uint32{volumeID},
+		Mode:      volume_server_pb.VolumeScrubMode_LOCAL,
+	})
+	if err != nil {
+		t.Fatalf("ScrubEcVolume LOCAL on healthy volume failed: %v", err)
+	}
+	if resp.GetTotalVolumes() != 1 {
+		t.Fatalf("expected total_volumes=1, got %d", resp.GetTotalVolumes())
+	}
+	if resp.GetTotalFiles() != 1 {
+		t.Fatalf("expected total_files=1, got %d", resp.GetTotalFiles())
+	}
+	if len(resp.GetBrokenVolumeIds()) != 0 {
+		t.Fatalf("expected no broken volumes, got %v: %v", resp.GetBrokenVolumeIds(), resp.GetDetails())
+	}
+	if len(resp.GetBrokenShardInfos()) != 0 {
+		t.Fatalf("expected no broken shards, got %v", resp.GetBrokenShardInfos())
+	}
+}
+
+func TestScrubEcVolumeLocalCorruptShard(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(212)
+	httpClient := framework.NewHTTPClient()
+	ecSetup(t, grpcClient, httpClient, clusterHarness.VolumeAdminURL(), volumeID)
+
+	// Corrupt shard 0 by truncating it.
+	framework.CorruptEcShardFile(t, clusterHarness.BaseDir(), volumeID, 0)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	resp, err := grpcClient.ScrubEcVolume(ctx, &volume_server_pb.ScrubEcVolumeRequest{
+		VolumeIds: []uint32{volumeID},
+		Mode:      volume_server_pb.VolumeScrubMode_LOCAL,
+	})
+	if err != nil {
+		t.Fatalf("ScrubEcVolume LOCAL on corrupt shard failed: %v", err)
+	}
+	if len(resp.GetBrokenVolumeIds()) == 0 {
+		t.Fatalf("expected broken volume after shard corruption")
+	}
+	if len(resp.GetBrokenShardInfos()) == 0 {
+		t.Fatalf("expected broken shard info after shard corruption")
+	}
+	// Verify all reported broken shards belong to the corrupted volume.
+	for _, si := range resp.GetBrokenShardInfos() {
+		if si.GetVolumeId() != volumeID {
+			t.Fatalf("broken shard info for unexpected volume %d, want %d", si.GetVolumeId(), volumeID)
+		}
+	}
+}
+
+func TestScrubEcVolumeAutoSelectWithEcPresent(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 volumeIDA = uint32(213)
+	const volumeIDB = uint32(214)
+	httpClient := framework.NewHTTPClient()
+	ecSetup(t, grpcClient, httpClient, clusterHarness.VolumeAdminURL(), volumeIDA)
+	ecSetup(t, grpcClient, httpClient, clusterHarness.VolumeAdminURL(), volumeIDB)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	// Auto-select (empty VolumeIds) should find both EC volumes.
+	resp, err := grpcClient.ScrubEcVolume(ctx, &volume_server_pb.ScrubEcVolumeRequest{
+		Mode: volume_server_pb.VolumeScrubMode_INDEX,
+	})
+	if err != nil {
+		t.Fatalf("ScrubEcVolume auto-select failed: %v", err)
+	}
+	if resp.GetTotalVolumes() < 2 {
+		t.Fatalf("expected at least 2 EC volumes via auto-select, got %d", resp.GetTotalVolumes())
+	}
+	if len(resp.GetBrokenVolumeIds()) != 0 {
+		t.Fatalf("expected no broken volumes, got %v", resp.GetBrokenVolumeIds())
+	}
+}
+
+func TestScrubEcVolumeUnsupportedMode(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(215)
+	httpClient := framework.NewHTTPClient()
+	ecSetup(t, grpcClient, httpClient, clusterHarness.VolumeAdminURL(), volumeID)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	_, err := grpcClient.ScrubEcVolume(ctx, &volume_server_pb.ScrubEcVolumeRequest{
+		VolumeIds: []uint32{volumeID},
+		Mode:      volume_server_pb.VolumeScrubMode(99),
+	})
+	if err == nil {
+		t.Fatalf("ScrubEcVolume should fail for unsupported mode")
+	}
+}
+
+func TestScrubEcVolumeIndexCorruptEcx(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(216)
+	framework.AllocateVolume(t, grpcClient, volumeID, "")
+
+	httpClient := framework.NewHTTPClient()
+	fid := framework.NewFileID(volumeID, 1, 0xABCD0001)
+	uploadResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid, []byte("ec-ecx-corrupt-test"))
+	_ = 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()
+
+	// Generate EC shards (this creates the .ecx file on disk).
+	_, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{
+		VolumeId:   volumeID,
+		Collection: "",
+	})
+	if err != nil {
+		t.Fatalf("VolumeEcShardsGenerate failed: %v", err)
+	}
+
+	// Corrupt the .ecx file BEFORE mounting, so the corrupted size is loaded.
+	framework.CorruptEcxFile(t, clusterHarness.BaseDir(), volumeID)
+
+	// Now mount shards - the ecx file size will reflect the corruption.
+	allShards := make([]uint32, erasure_coding.TotalShardsCount)
+	for i := range allShards {
+		allShards[i] = uint32(i)
+	}
+	_, err = grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{
+		VolumeId:   volumeID,
+		Collection: "",
+		ShardIds:   allShards,
+	})
+	if err != nil {
+		t.Fatalf("VolumeEcShardsMount failed: %v", err)
+	}
+
+	resp, err := grpcClient.ScrubEcVolume(ctx, &volume_server_pb.ScrubEcVolumeRequest{
+		VolumeIds: []uint32{volumeID},
+		Mode:      volume_server_pb.VolumeScrubMode_INDEX,
+	})
+	if err != nil {
+		t.Fatalf("ScrubEcVolume INDEX on corrupt ecx failed: %v", err)
+	}
+	if len(resp.GetBrokenVolumeIds()) == 0 {
+		t.Fatalf("expected broken volume after ECX corruption")
+	}
+}

weed/admin/static/js/admin.js

@@ -1267,7 +1267,7 @@ async function submitUploadFile() {
         });
 
         // Send request
-        xhr.open('POST', '/api/files/upload');
+        xhr.open('POST', basePath('/api/files/upload'));
         xhr.send(formData);
 
     } catch (error) {
@@ -1320,7 +1320,7 @@ function exportFileList() {
 // Download file
 function downloadFile(filePath) {
     // Create download link using admin API
-    const downloadUrl = `/api/files/download?path=${encodeURIComponent(filePath)}`;
+    const downloadUrl = basePath(`/api/files/download?path=${encodeURIComponent(filePath)}`);
     window.open(downloadUrl, '_blank');
 }
 
@@ -1786,7 +1786,7 @@ function createFileViewerContent(file, content) {
     if (file.mime.startsWith('image/')) {
         return `
             <div class="text-center">
-                <img src="/api/files/download?path=${encodeURIComponent(file.full_path)}" 
+                <img src="${basePath('/api/files/download?path=' + encodeURIComponent(file.full_path))}"
                      class="img-fluid" alt="${file.name}" style="max-height: 500px;">
             </div>
         `;
@@ -1804,7 +1804,7 @@ function createFileViewerContent(file, content) {
     } else if (file.mime === 'application/pdf') {
         return `
             <div class="text-center">
-                <embed src="/api/files/download?path=${encodeURIComponent(file.full_path)}" 
+                <embed src="${basePath('/api/files/download?path=' + encodeURIComponent(file.full_path))}"
                        type="application/pdf" width="100%" height="500px">
             </div>
         `;

weed/command/admin.go

@@ -392,7 +392,19 @@ func startAdminServer(ctx context.Context, options AdminOptions, enableUI bool,
 	addr := fmt.Sprintf(":%d", *options.port)
 	var handler http.Handler = r
 	if urlPrefix != "" {
-		handler = http.StripPrefix(urlPrefix, r)
+		stripped := http.StripPrefix(urlPrefix, r)
+		handler = http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+			// Redirect /prefix (no trailing slash) to /prefix/
+			if req.URL.Path == urlPrefix {
+				target := urlPrefix + "/"
+				if req.URL.RawQuery != "" {
+					target += "?" + req.URL.RawQuery
+				}
+				http.Redirect(w, req, target, http.StatusFound)
+				return
+			}
+			stripped.ServeHTTP(w, req)
+		})
 	}
 	server := &http.Server{
 		Addr:    addr,

weed/command/filer_sync.go

@@ -15,12 +15,14 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/replication"
 	"github.com/seaweedfs/seaweedfs/weed/replication/sink"
+	"github.com/seaweedfs/seaweedfs/weed/operation"
 	"github.com/seaweedfs/seaweedfs/weed/replication/sink/filersink"
 	"github.com/seaweedfs/seaweedfs/weed/replication/source"
 	"github.com/seaweedfs/seaweedfs/weed/security"
 	statsCollect "github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"github.com/seaweedfs/seaweedfs/weed/util/grace"
+	util_http_client "github.com/seaweedfs/seaweedfs/weed/util/http/client"
 	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
 	"google.golang.org/grpc"
 )
@@ -164,6 +166,21 @@ func runFilerSynchronize(cmd *Command, args []string) bool {
 		}
 	}
 
+	// per-cluster HTTPS clients for volume server connections
+	var httpClientA, httpClientB *util_http_client.HTTPClient
+	if *syncOptions.aSecurity != "" {
+		var err error
+		if httpClientA, err = security.LoadHTTPClientFromFile(*syncOptions.aSecurity); err != nil {
+			glog.Fatalf("load HTTPS client config for filer A: %v", err)
+		}
+	}
+	if *syncOptions.bSecurity != "" {
+		var err error
+		if httpClientB, err = security.LoadHTTPClientFromFile(*syncOptions.bSecurity); err != nil {
+			glog.Fatalf("load HTTPS client config for filer B: %v", err)
+		}
+	}
+
 	grace.SetupProfiling(*syncCpuProfile, *syncMemProfile)
 
 	filerA := pb.ServerAddress(*syncOptions.filerA)
@@ -238,7 +255,9 @@ func runFilerSynchronize(cmd *Command, args []string) bool {
 				*syncOptions.bDoDeleteFiles,
 				aFilerSignature,
 				bFilerSignature,
-				&syncStateA2B)
+				&syncStateA2B,
+				httpClientA,
+				httpClientB)
 			if err != nil {
 				glog.Errorf("sync from %s to %s: %v", *syncOptions.filerA, *syncOptions.filerB, err)
 				time.Sleep(1747 * time.Millisecond)
@@ -279,7 +298,9 @@ func runFilerSynchronize(cmd *Command, args []string) bool {
 					*syncOptions.aDoDeleteFiles,
 					bFilerSignature,
 					aFilerSignature,
-					&syncStateB2A)
+					&syncStateB2A,
+					httpClientB,
+					httpClientA)
 				if err != nil {
 					glog.Errorf("sync from %s to %s: %v", *syncOptions.filerB, *syncOptions.filerA, err)
 					time.Sleep(2147 * time.Millisecond)
@@ -308,7 +329,8 @@ func initOffsetFromTsMs(grpcDialOption grpc.DialOption, targetFiler pb.ServerAdd
 }
 
 func doSubscribeFilerMetaChanges(clientId int32, clientEpoch int32, sourceGrpcDialOption grpc.DialOption, sourceFiler pb.ServerAddress, sourcePath string, sourceExcludePaths []string, sourceReadChunkFromFiler bool, targetGrpcDialOption grpc.DialOption, targetFiler pb.ServerAddress, targetPath string,
-	replicationStr, collection string, ttlSec int, sinkWriteChunkByFiler bool, diskType string, debug bool, concurrency int, chunkConcurrency int, doDeleteFiles bool, sourceFilerSignature int32, targetFilerSignature int32, statePtr *atomic.Pointer[syncState]) error {
+	replicationStr, collection string, ttlSec int, sinkWriteChunkByFiler bool, diskType string, debug bool, concurrency int, chunkConcurrency int, doDeleteFiles bool, sourceFilerSignature int32, targetFilerSignature int32, statePtr *atomic.Pointer[syncState],
+	sourceHttpClient *util_http_client.HTTPClient, sinkHttpClient *util_http_client.HTTPClient) error {
 
 	// if first time, start from now
 	// if has previously synced, resume from that point of time
@@ -323,9 +345,15 @@ func doSubscribeFilerMetaChanges(clientId int32, clientEpoch int32, sourceGrpcDi
 	filerSource := &source.FilerSource{}
 	filerSource.DoInitialize(sourceFiler.ToHttpAddress(), sourceFiler.ToGrpcAddress(), sourcePath, sourceReadChunkFromFiler)
 	filerSource.SetGrpcDialOption(sourceGrpcDialOption)
+	if sourceHttpClient != nil {
+		filerSource.SetHttpClient(sourceHttpClient)
+	}
 	filerSink := &filersink.FilerSink{}
 	filerSink.DoInitialize(targetFiler.ToHttpAddress(), targetFiler.ToGrpcAddress(), targetPath, replicationStr, collection, ttlSec, diskType, targetGrpcDialOption, sinkWriteChunkByFiler)
 	filerSink.SetChunkConcurrency(chunkConcurrency)
+	if sinkHttpClient != nil {
+		filerSink.SetUploader(operation.NewUploaderWithHttpClient(sinkHttpClient))
+	}
 	filerSink.SetSourceFiler(filerSource)
 
 	persistEventFn := genProcessFunction(sourcePath, targetPath, sourceExcludePaths, nil, nil, nil, filerSink, doDeleteFiles, debug)

weed/command/scaffold/filer.toml

@@ -13,6 +13,9 @@
 # recursive_delete will delete all sub folders and files, similar to "rm -Rf"
 recursive_delete = false
 #max_file_name_length = 255
+# for S3: how long to wait before deleting an empty folder.
+# increase this if using tools like Spark that create temporary directories.
+#s3.empty_folder_cleanup_delay = "2m"
 
 ####################################################
 # The following are filer store options

weed/command/shell.go

@@ -2,6 +2,7 @@ package command
 
 import (
 	"fmt"
+	"os"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 
@@ -60,7 +61,7 @@ func runShell(command *Command, args []string) bool {
 		filerAddress = viper.GetString("cluster." + cluster + ".filer")
 	}
 	shellOptions.FilerAddress = pb.ServerAddress(filerAddress)
-	fmt.Printf("master: %s filer: %s\n", *shellOptions.Masters, shellOptions.FilerAddress)
+	fmt.Fprintf(os.Stderr, "master: %s filer: %s\n", *shellOptions.Masters, shellOptions.FilerAddress)
 
 	shell.RunShell(shellOptions)
 

weed/filer/empty_folder_cleanup/cleanup_queue.go

@@ -153,6 +153,27 @@ func (q *CleanupQueue) Pop() (string, string, bool) {
 	return item.folder, item.triggeredBy, true
 }
 
+// PopOlderThan removes and returns the oldest folder only if it has been in the queue
+// for longer than the specified duration. Returns empty string and false if no item qualifies.
+func (q *CleanupQueue) PopOlderThan(minAge time.Duration) (string, string, bool) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	front := q.items.Front()
+	if front == nil {
+		return "", "", false
+	}
+
+	item := front.Value.(*queueItem)
+	if time.Since(item.queueTime) <= minAge {
+		return "", "", false
+	}
+
+	q.items.Remove(front)
+	delete(q.itemsMap, item.folder)
+	return item.folder, item.triggeredBy, true
+}
+
 // Peek returns the oldest folder without removing it.
 // Returns the folder and queue time if available, or empty values if queue is empty.
 func (q *CleanupQueue) Peek() (folder string, triggeredBy string, queueTime time.Time, ok bool) {

weed/filer/empty_folder_cleanup/empty_folder_cleaner.go

@@ -18,8 +18,8 @@ const (
 	DefaultMaxCountCheck  = 1000
 	DefaultCacheExpiry    = 5 * time.Minute
 	DefaultQueueMaxSize   = 1000
-	DefaultQueueMaxAge    = 5 * time.Second
-	DefaultProcessorSleep = 10 * time.Second // How often to check queue
+	DefaultQueueMaxAge    = 2 * time.Minute
+	DefaultProcessorSleep = 30 * time.Second // How often to check queue
 )
 
 // FilerOperations defines the filer operations needed by EmptyFolderCleaner
@@ -70,15 +70,20 @@ type EmptyFolderCleaner struct {
 	stopCh  chan struct{}
 }
 
-// NewEmptyFolderCleaner creates a new EmptyFolderCleaner
-func NewEmptyFolderCleaner(filer FilerOperations, lockRing *lock_manager.LockRing, host pb.ServerAddress, bucketPath string) *EmptyFolderCleaner {
+// NewEmptyFolderCleaner creates a new EmptyFolderCleaner.
+// cleanupDelay controls how long an empty folder must remain in the queue before deletion.
+// If zero, DefaultQueueMaxAge is used.
+func NewEmptyFolderCleaner(filer FilerOperations, lockRing *lock_manager.LockRing, host pb.ServerAddress, bucketPath string, cleanupDelay time.Duration) *EmptyFolderCleaner {
+	if cleanupDelay <= 0 {
+		cleanupDelay = DefaultQueueMaxAge
+	}
 	efc := &EmptyFolderCleaner{
 		filer:                 filer,
 		lockRing:              lockRing,
 		host:                  host,
 		folderCounts:          make(map[string]*folderState),
 		bucketCleanupPolicies: make(map[string]*bucketCleanupPolicyState),
-		cleanupQueue:          NewCleanupQueue(DefaultQueueMaxSize, DefaultQueueMaxAge),
+		cleanupQueue:          NewCleanupQueue(DefaultQueueMaxSize, cleanupDelay),
 		maxCountCheck:         DefaultMaxCountCheck,
 		cacheExpiry:           DefaultCacheExpiry,
 		processorSleep:        DefaultProcessorSleep,
@@ -207,27 +212,22 @@ func (efc *EmptyFolderCleaner) cleanupProcessor() {
 
 // processCleanupQueue processes items from the cleanup queue
 func (efc *EmptyFolderCleaner) processCleanupQueue() {
-	// Check if we should process
-	if !efc.cleanupQueue.ShouldProcess() {
-		if efc.cleanupQueue.Len() > 0 {
-			glog.Infof("EmptyFolderCleaner: pending queue not processed yet (len=%d, oldest_age=%v, max_size=%d, max_age=%v)",
-				efc.cleanupQueue.Len(), efc.cleanupQueue.OldestAge(), efc.cleanupQueue.maxSize, efc.cleanupQueue.maxAge)
-		}
+	if efc.cleanupQueue.Len() == 0 {
 		return
 	}
 
-	glog.V(3).Infof("EmptyFolderCleaner: processing cleanup queue (len=%d, age=%v)",
+	glog.V(3).Infof("EmptyFolderCleaner: processing cleanup queue (len=%d, oldest_age=%v)",
 		efc.cleanupQueue.Len(), efc.cleanupQueue.OldestAge())
 
-	// Process all items that are ready
-	for efc.cleanupQueue.Len() > 0 {
+	// Only process items that have been queued longer than maxAge
+	for {
 		// Check if still enabled
 		if !efc.IsEnabled() {
 			return
 		}
 
-		// Pop the oldest item
-		folder, triggeredBy, ok := efc.cleanupQueue.Pop()
+		// Only pop items old enough — newer items stay in the queue
+		folder, triggeredBy, ok := efc.cleanupQueue.PopOlderThan(efc.cleanupQueue.maxAge)
 		if !ok {
 			break
 		}
@@ -322,9 +322,18 @@ func (efc *EmptyFolderCleaner) executeCleanup(folder string, triggeredBy string)
 	delete(efc.folderCounts, folder)
 	efc.mu.Unlock()
 
-	// Note: No need to recursively check parent folder here.
-	// The deletion of this folder will generate a metadata event,
-	// which will trigger OnDeleteEvent for the parent folder.
+	// After deleting this folder, immediately try to clean the parent.
+	// Relying solely on cascading metadata events would re-enter the full
+	// delay queue for each ancestor level, causing multi-minute cascading
+	// waits (e.g. 3 levels × 2m = 6m+).  Instead, walk up eagerly.
+	parentDir, _ := util.FullPath(folder).DirAndName()
+	if parentDir != "" && parentDir != folder &&
+		efc.bucketPath != "" && isUnderBucketPath(parentDir, efc.bucketPath) {
+		// Remove any pending queue entry for the parent so we don't
+		// double-process it later from a stale event.
+		efc.cleanupQueue.Remove(parentDir)
+		efc.executeCleanup(parentDir, triggeredBy)
+	}
 }
 
 // countItems counts items in a folder (up to maxCountCheck)

weed/filer/empty_folder_cleanup/empty_folder_cleaner_test.go

@@ -655,7 +655,7 @@ func TestEmptyFolderCleaner_queueFIFOOrder(t *testing.T) {
 	cleaner.Stop()
 }
 
-func TestEmptyFolderCleaner_processCleanupQueue_drainsAllOnceTriggered(t *testing.T) {
+func TestEmptyFolderCleaner_processCleanupQueue_onlyProcessesAgedItems(t *testing.T) {
 	lockRing := lock_manager.NewLockRing(5 * time.Second)
 	lockRing.SetSnapshot([]pb.ServerAddress{"filer1:8888"}, 0)
 
@@ -670,6 +670,7 @@ func TestEmptyFolderCleaner_processCleanupQueue_drainsAllOnceTriggered(t *testin
 		},
 	}
 
+	maxAge := 100 * time.Millisecond
 	cleaner := &EmptyFolderCleaner{
 		filer:          mock,
 		lockRing:       lockRing,
@@ -677,25 +678,27 @@ func TestEmptyFolderCleaner_processCleanupQueue_drainsAllOnceTriggered(t *testin
 		bucketPath:     "/buckets",
 		enabled:        true,
 		folderCounts:   make(map[string]*folderState),
-		cleanupQueue:   NewCleanupQueue(2, time.Hour),
+		cleanupQueue:   NewCleanupQueue(1000, maxAge),
 		maxCountCheck:  1000,
 		cacheExpiry:    time.Minute,
 		processorSleep: time.Second,
 		stopCh:         make(chan struct{}),
 	}
 
-	now := time.Now()
-	cleaner.cleanupQueue.Add("/buckets/test/folder1", "i1", now)
-	cleaner.cleanupQueue.Add("/buckets/test/folder2", "i2", now.Add(time.Millisecond))
-	cleaner.cleanupQueue.Add("/buckets/test/folder3", "i3", now.Add(2*time.Millisecond))
+	// Add old items (well past maxAge) and a fresh item
+	old := time.Now().Add(-time.Second)
+	cleaner.cleanupQueue.Add("/buckets/test/folder1", "i1", old)
+	cleaner.cleanupQueue.Add("/buckets/test/folder2", "i2", old.Add(time.Millisecond))
+	cleaner.cleanupQueue.Add("/buckets/test/folder3", "i3", time.Now()) // fresh, should NOT be processed
 
 	cleaner.processCleanupQueue()
 
-	if got := cleaner.cleanupQueue.Len(); got != 0 {
-		t.Fatalf("expected queue to be drained, got len=%d", got)
+	// Only the two old items should have been processed
+	if len(deleted) != 2 {
+		t.Fatalf("expected 2 deleted folders (aged items only), got %d: %v", len(deleted), deleted)
 	}
-	if len(deleted) != 3 {
-		t.Fatalf("expected 3 deleted folders, got %d", len(deleted))
+	if got := cleaner.cleanupQueue.Len(); got != 1 {
+		t.Fatalf("expected 1 item remaining in queue, got %d", got)
 	}
 }
 

weed/filer/filer.go

@@ -61,7 +61,8 @@ type Filer struct {
 	MaxFilenameLength   uint32
 	deletionQuit        chan struct{}
 	DeletionRetryQueue  *DeletionRetryQueue
-	EmptyFolderCleaner  *empty_folder_cleanup.EmptyFolderCleaner
+	EmptyFolderCleaner       *empty_folder_cleanup.EmptyFolderCleaner
+	EmptyFolderCleanupDelay time.Duration
 }
 
 func NewFiler(masters pb.ServerDiscovery, grpcDialOption grpc.DialOption, filerHost pb.ServerAddress, filerGroup string, collection string, replication string, dataCenter string, maxFilenameLength uint32, notifyFn func()) *Filer {
@@ -123,7 +124,7 @@ func (f *Filer) AggregateFromPeers(self pb.ServerAddress, existingNodes []*maste
 	glog.V(0).Infof("%s aggregate from peers %+v", self, snapshot)
 
 	// Initialize the empty folder cleaner using the same LockRing as Dlm for consistent hashing
-	f.EmptyFolderCleaner = empty_folder_cleanup.NewEmptyFolderCleaner(f, f.Dlm.LockRing, self, f.DirBucketsPath)
+	f.EmptyFolderCleaner = empty_folder_cleanup.NewEmptyFolderCleaner(f, f.Dlm.LockRing, self, f.DirBucketsPath, f.EmptyFolderCleanupDelay)
 
 	f.MetaAggregator = NewMetaAggregator(f, self, f.GrpcDialOption)
 	f.MasterClient.SetOnPeerUpdateFn(func(update *master_pb.ClusterNodeUpdate, startFrom time.Time) {

weed/filer/meta_aggregator.go

@@ -2,6 +2,7 @@ package filer
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"strings"
@@ -139,16 +140,37 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 	if peerSignature != f.Signature {
 		if prevTsNs, err := ma.readOffset(f, peer, peerSignature); err == nil {
 			lastTsNs = prevTsNs
-			defer func(prevTsNs int64) {
-				if lastTsNs != prevTsNs && lastTsNs != lastPersistTime.UnixNano() {
-					if err := ma.updateOffset(f, peer, peerSignature, lastTsNs); err == nil {
-						glog.V(0).Infof("last sync time with %s at %v (%d)", peer, time.Unix(0, lastTsNs), lastTsNs)
-					} else {
-						glog.Errorf("failed to save last sync time with %s at %v (%d)", peer, time.Unix(0, lastTsNs), lastTsNs)
-					}
-				}
-			}(prevTsNs)
+		} else if errors.Is(err, ErrKvNotFound) {
+			// No stored offset — this is the first time connecting to this peer.
+			// Traverse the peer's full metadata tree so we get pre-existing data.
+			// Record time before traversal and subtract a safety margin to
+			// account for clock skew between this filer and the peer. Any
+			// duplicate events replayed during the overlap are harmless since
+			// Replay does upserts. We use wall-clock time (same domain as the
+			// metadata stream TsNs) rather than entry Mtime which is a
+			// different concept and can be set to arbitrary values.
+			preTraverseTime := time.Now()
+			glog.V(0).Infof("no previous offset for peer %s, starting full metadata sync", peer)
+			if traverseErr := ma.traversePeerMetadata(f, peer); traverseErr != nil {
+				return lastTsNs, fmt.Errorf("initial metadata sync from %s: %v", peer, traverseErr)
+			}
+			lastTsNs = preTraverseTime.Add(-time.Minute).UnixNano()
+			if err := ma.updateOffset(f, peer, peerSignature, lastTsNs); err != nil {
+				return lastTsNs, fmt.Errorf("save bootstrap offset for peer %s: %w", peer, err)
+			}
+			glog.V(0).Infof("completed full metadata sync from peer %s, will stream changes from %v", peer, time.Unix(0, lastTsNs))
+		} else {
+			return lastTsNs, fmt.Errorf("read offset for peer %s: %w", peer, err)
 		}
+		defer func(prevTsNs int64) {
+			if lastTsNs != prevTsNs && lastTsNs != lastPersistTime.UnixNano() {
+				if err := ma.updateOffset(f, peer, peerSignature, lastTsNs); err == nil {
+					glog.V(0).Infof("last sync time with %s at %v (%d)", peer, time.Unix(0, lastTsNs), lastTsNs)
+				} else {
+					glog.Errorf("failed to save last sync time with %s at %v (%d)", peer, time.Unix(0, lastTsNs), lastTsNs)
+				}
+			}
+		}(lastTsNs)
 
 		glog.V(0).Infof("follow peer: %v, last %v (%d)", peer, time.Unix(0, lastTsNs), lastTsNs)
 		var counter int64
@@ -279,6 +301,59 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 	return lastTsNs, err
 }
 
+// traversePeerMetadata does a full BFS traversal of a peer filer's metadata
+// and inserts all entries into the local store. This is used when a filer
+// connects to a peer for the first time and needs to bootstrap pre-existing data.
+func (ma *MetaAggregator) traversePeerMetadata(f *Filer, peer pb.ServerAddress) error {
+	return pb.WithFilerClient(true, 0, peer, ma.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		stream, err := client.TraverseBfsMetadata(ctx, &filer_pb.TraverseBfsMetadataRequest{
+			Directory:        "/",
+			ExcludedPrefixes: []string{SystemLogDir},
+		})
+		if err != nil {
+			return fmt.Errorf("traverse bfs metadata: %w", err)
+		}
+		var count int64
+		for {
+			resp, recvErr := stream.Recv()
+			if recvErr == io.EOF {
+				break
+			}
+			if recvErr != nil {
+				return fmt.Errorf("traverse bfs metadata recv: %w", recvErr)
+			}
+			if resp.Entry == nil {
+				continue
+			}
+			fullpath := util.Join(resp.Directory, resp.Entry.Name)
+			entry := FromPbEntry(resp.Directory, resp.Entry)
+			if insertErr := f.Store.InsertEntry(context.Background(), entry); insertErr != nil {
+				// Entry may already exist (root dir, or partial previous bootstrap).
+				existing, findErr := f.Store.FindEntry(context.Background(), entry.FullPath)
+				if findErr != nil {
+					return fmt.Errorf("insert entry %s: %w", fullpath, insertErr)
+				}
+				// Only overwrite if the peer's entry is newer.
+				if entry.Attr.Mtime.After(existing.Attr.Mtime) {
+					if updateErr := f.Store.UpdateEntry(context.Background(), entry); updateErr != nil {
+						return fmt.Errorf("update entry %s: %w", fullpath, updateErr)
+					}
+				} else {
+					glog.V(1).Infof("skip older peer entry %s (peer mtime %v <= local mtime %v)", fullpath, entry.Attr.Mtime, existing.Attr.Mtime)
+				}
+			}
+			count++
+			if count%10000 == 0 {
+				glog.V(0).Infof("synced %d entries from peer %s", count, peer)
+			}
+		}
+		glog.V(0).Infof("synced %d entries total from peer %s", count, peer)
+		return nil
+	})
+}
+
 func (ma *MetaAggregator) readFilerStoreSignature(peer pb.ServerAddress) (sig int32, err error) {
 	err = pb.WithFilerClient(false, 0, peer, ma.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
 		resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
@@ -308,7 +383,7 @@ func (ma *MetaAggregator) readOffset(f *Filer, peer pb.ServerAddress, peerSignat
 	value, err := f.Store.KvGet(context.Background(), key)
 
 	if err != nil {
-		return 0, fmt.Errorf("readOffset %s : %v", peer, err)
+		return 0, fmt.Errorf("readOffset %s : %w", peer, err)
 	}
 
 	lastTsNs = int64(util.BytesToUint64(value))

weed/filer/redis2/universal_redis_store.go

@@ -197,15 +197,18 @@ func (store *UniversalRedis2Store) ListDirectoryEntries(ctx context.Context, dir
 	}
 
 	// fetch entry meta
+	var entry *filer.Entry
 	for _, fileName := range members {
 		path := util.NewFullPath(string(dirPath), fileName)
-		entry, err := store.FindEntry(ctx, path)
+		entry, err = store.FindEntry(ctx, path)
 		lastFileName = fileName
 		if err != nil {
 			glog.V(0).InfofCtx(ctx, "list %s : %v", path, err)
 			if err == filer_pb.ErrNotFound {
+				err = nil
 				continue
 			}
+			break
 		} else {
 			if entry.TtlSec > 0 {
 				if entry.Attr.Crtime.Add(time.Duration(entry.TtlSec) * time.Second).Before(time.Now()) {

weed/iam/policy/aws_iam_compliance_test.go

@@ -1,8 +1,10 @@
 package policy
 
 import (
+	"strings"
 	"testing"
 
+	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -286,7 +288,7 @@ func TestAWSWildcardMatch(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			result := AwsWildcardMatch(tt.pattern, tt.value)
+			result := wildcard.MatchesWildcard(strings.ToLower(tt.pattern), strings.ToLower(tt.value))
 			assert.Equal(t, tt.expected, result, "AWS wildcard match should match expected")
 		})
 	}

weed/iam/policy/policy_engine.go

@@ -5,12 +5,12 @@ import (
 	"encoding/json"
 	"fmt"
 	"net"
-	"path/filepath"
 	"regexp"
 	"strconv"
 	"strings"
-	"sync"
 	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
 )
 
 // Effect represents the policy evaluation result
@@ -21,10 +21,7 @@ const (
 	EffectDeny  Effect = "Deny"
 )
 
-// Package-level regex cache for performance optimization
 var (
-	regexCache            = make(map[string]*regexp.Regexp)
-	regexCacheMu          sync.RWMutex
 	policyVariablePattern = regexp.MustCompile(`\$\{([^}]+)\}`)
 	safePolicyVariables   = map[string]bool{
 		// AWS standard identity variables
@@ -1064,8 +1061,7 @@ func (e *PolicyEngine) EvaluateStringCondition(block map[string]interface{}, eva
 				for _, expected := range expectedStrings {
 					expandedExpected := expandPolicyVariables(expected, evalCtx)
 					if useWildcard {
-						// Use filepath.Match for case-sensitive wildcard matching, as required by StringLike
-						if matched, _ := filepath.Match(expandedExpected, contextValue); matched {
+						if wildcard.MatchesWildcard(expandedExpected, contextValue) {
 							contextValueMatchedSet = true
 							break
 						}
@@ -1106,13 +1102,11 @@ func (e *PolicyEngine) EvaluateStringCondition(block map[string]interface{}, eva
 				for _, expected := range expectedStrings {
 					expandedExpected := expandPolicyVariables(expected, evalCtx)
 					if useWildcard {
-						// Use filepath.Match for case-sensitive wildcard matching, as required by StringLike
-						if matched, _ := filepath.Match(expandedExpected, contextValue); matched {
+						if wildcard.MatchesWildcard(expandedExpected, contextValue) {
 							contextValueMatchedSet = true
 							break
 						}
 					} else {
-						// For StringEquals/StringNotEquals, also support policy variables but be case-sensitive
 						if expandedExpected == contextValue {
 							contextValueMatchedSet = true
 							break
@@ -1229,7 +1223,7 @@ func awsIAMMatch(pattern, value string, evalCtx *EvaluationContext) bool {
 
 	// Step 4: Handle AWS-style wildcards (case-insensitive)
 	if strings.Contains(expandedPattern, "*") || strings.Contains(expandedPattern, "?") {
-		return AwsWildcardMatch(expandedPattern, value)
+		return wildcard.MatchesWildcard(strings.ToLower(expandedPattern), strings.ToLower(value))
 	}
 
 	return false
@@ -1265,44 +1259,6 @@ func expandPolicyVariables(pattern string, evalCtx *EvaluationContext) string {
 	return result
 }
 
-// AwsWildcardMatch performs case-insensitive wildcard matching like AWS IAM
-func AwsWildcardMatch(pattern, value string) bool {
-	// Create regex pattern key for caching
-	// First escape all regex metacharacters, then replace wildcards
-	regexPattern := regexp.QuoteMeta(pattern)
-	regexPattern = strings.ReplaceAll(regexPattern, "\\*", ".*")
-	regexPattern = strings.ReplaceAll(regexPattern, "\\?", ".")
-	regexPattern = "^" + regexPattern + "$"
-	regexKey := "(?i)" + regexPattern
-
-	// Try to get compiled regex from cache
-	regexCacheMu.RLock()
-	regex, found := regexCache[regexKey]
-	regexCacheMu.RUnlock()
-
-	if !found {
-		// Compile and cache the regex
-		compiledRegex, err := regexp.Compile(regexKey)
-		if err != nil {
-			// Fallback to simple case-insensitive comparison if regex fails
-			return strings.EqualFold(pattern, value)
-		}
-
-		// Store in cache with write lock
-		regexCacheMu.Lock()
-		// Double-check in case another goroutine added it
-		if existingRegex, exists := regexCache[regexKey]; exists {
-			regex = existingRegex
-		} else {
-			regexCache[regexKey] = compiledRegex
-			regex = compiledRegex
-		}
-		regexCacheMu.Unlock()
-	}
-
-	return regex.MatchString(value)
-}
-
 // evaluateStringConditionIgnoreCase evaluates string conditions with case insensitivity
 func (e *PolicyEngine) evaluateStringConditionIgnoreCase(block map[string]interface{}, evalCtx *EvaluationContext, shouldMatch bool, useWildcard bool, forAllValues bool) bool {
 	for key, expectedValues := range block {
@@ -1347,7 +1303,7 @@ func (e *PolicyEngine) evaluateStringConditionIgnoreCase(block map[string]interf
 				case string:
 					expandedPattern := expandPolicyVariables(v, evalCtx)
 					if useWildcard {
-						if AwsWildcardMatch(expandedPattern, ctxStr) {
+						if wildcard.MatchesWildcard(strings.ToLower(expandedPattern), strings.ToLower(ctxStr)) {
 							itemMatchedSet = true
 						}
 					} else {
@@ -1369,7 +1325,7 @@ func (e *PolicyEngine) evaluateStringConditionIgnoreCase(block map[string]interf
 					for _, valStr := range slice {
 						expandedPattern := expandPolicyVariables(valStr, evalCtx)
 						if useWildcard {
-							if AwsWildcardMatch(expandedPattern, ctxStr) {
+							if wildcard.MatchesWildcard(strings.ToLower(expandedPattern), strings.ToLower(ctxStr)) {
 								itemMatchedSet = true
 								break
 							}
@@ -1409,7 +1365,7 @@ func (e *PolicyEngine) evaluateStringConditionIgnoreCase(block map[string]interf
 				case string:
 					expandedPattern := expandPolicyVariables(v, evalCtx)
 					if useWildcard {
-						if AwsWildcardMatch(expandedPattern, ctxStr) {
+						if wildcard.MatchesWildcard(strings.ToLower(expandedPattern), strings.ToLower(ctxStr)) {
 							itemMatchedSet = true
 						}
 					} else {
@@ -1431,7 +1387,7 @@ func (e *PolicyEngine) evaluateStringConditionIgnoreCase(block map[string]interf
 					for _, valStr := range slice {
 						expandedPattern := expandPolicyVariables(valStr, evalCtx)
 						if useWildcard {
-							if AwsWildcardMatch(expandedPattern, ctxStr) {
+							if wildcard.MatchesWildcard(strings.ToLower(expandedPattern), strings.ToLower(ctxStr)) {
 								itemMatchedSet = true
 								break
 							}

weed/iam/providers/provider.go

@@ -4,10 +4,11 @@ import (
 	"context"
 	"fmt"
 	"net/mail"
+	"strings"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
-	"github.com/seaweedfs/seaweedfs/weed/iam/policy"
+	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
 )
 
 // IdentityProvider defines the interface for external identity providers
@@ -225,7 +226,7 @@ func (r *MappingRule) Matches(claims *TokenClaims) bool {
 // matchValue checks if a value matches the rule value (with wildcard support)
 // Uses AWS IAM-compliant case-insensitive wildcard matching for consistency with policy engine
 func (r *MappingRule) matchValue(value string) bool {
-	matched := policy.AwsWildcardMatch(r.Value, value)
+	matched := wildcard.MatchesWildcard(strings.ToLower(r.Value), strings.ToLower(value))
 	glog.V(3).Infof("AWS IAM pattern match result: '%s' matches '%s' = %t", value, r.Value, matched)
 	return matched
 }

weed/operation/upload_content.go

@@ -135,6 +135,15 @@ func newUploader(httpClient HTTPClient) *Uploader {
 	}
 }
 
+// NewUploaderWithHttpClient creates an Uploader that uses the provided HTTP
+// client instead of the global one. This is used by filer.sync to upload to
+// remote clusters that use different TLS certificates.
+func NewUploaderWithHttpClient(httpClient HTTPClient) *Uploader {
+	return &Uploader{
+		httpClient: httpClient,
+	}
+}
+
 func (uploader *Uploader) uploadWithRetryData(assignFn func() (fileId string, host string, auth security.EncodedJwt, err error), uploadOption *UploadOption, genFileUrlFn func(host, fileId string) string, data []byte) (fileId string, uploadResult *UploadResult, err error) {
 	doUploadFunc := func() error {
 		var host string

weed/replication/sink/filersink/fetch_write.go

@@ -178,7 +178,7 @@ func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk
 }
 
 func (fs *FilerSink) uploadManifestChunk(path string, sourceMtime int64, sourceFileId string, manifestData []byte) (fileId string, err error) {
-	uploader, err := operation.NewUploader()
+	uploader, err := fs.getUploader()
 	if err != nil {
 		glog.V(0).Infof("upload manifest data %v: %v", sourceFileId, err)
 		return "", fmt.Errorf("upload manifest data: %w", err)
@@ -235,7 +235,7 @@ func (fs *FilerSink) uploadManifestChunk(path string, sourceMtime int64, sourceF
 }
 
 func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, sourceMtime int64) (fileId string, err error) {
-	uploader, err := operation.NewUploader()
+	uploader, err := fs.getUploader()
 	if err != nil {
 		glog.V(0).Infof("upload source data %v: %v", sourceChunk.GetFileIdString(), err)
 		return "", fmt.Errorf("upload data: %w", err)

weed/replication/sink/filersink/filer_sink.go

@@ -6,6 +6,7 @@ import (
 	"math"
 	"sync"
 
+	"github.com/seaweedfs/seaweedfs/weed/operation"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/wdclient"
 
@@ -50,6 +51,7 @@ type FilerSink struct {
 	executor          *util.LimitedConcurrentExecutor
 	signature         int32
 	activeTransfers   sync.Map // chunkFileId -> *ChunkTransferStatus
+	uploader          *operation.Uploader
 }
 
 func init() {
@@ -88,6 +90,21 @@ func (fs *FilerSink) SetSourceFiler(s *source.FilerSource) {
 	fs.filerSource = s
 }
 
+// SetUploader sets a custom uploader for this sink, used when the target
+// cluster requires different TLS certificates than the global config.
+// Must be called during initialization, before any replication goroutines
+// start, since it writes fs.uploader without synchronization.
+func (fs *FilerSink) SetUploader(uploader *operation.Uploader) {
+	fs.uploader = uploader
+}
+
+func (fs *FilerSink) getUploader() (*operation.Uploader, error) {
+	if fs.uploader != nil {
+		return fs.uploader, nil
+	}
+	return operation.NewUploader()
+}
+
 func (fs *FilerSink) DoInitialize(address, grpcAddress string, dir string,
 	replication string, collection string, ttlSec int, diskType string, grpcDialOption grpc.DialOption, writeChunkByFiler bool) (err error) {
 	fs.address = address

weed/replication/source/filer_source.go

@@ -15,6 +15,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
+	util_http_client "github.com/seaweedfs/seaweedfs/weed/util/http/client"
 )
 
 type FilerSource struct {
@@ -25,6 +26,7 @@ type FilerSource struct {
 	proxyByFiler   bool
 	dataCenter     string
 	signature      int32
+	httpClient     *util_http_client.HTTPClient
 }
 
 func (fs *FilerSource) Initialize(configuration util.Configuration, prefix string) error {
@@ -54,6 +56,10 @@ func (fs *FilerSource) SetGrpcDialOption(option grpc.DialOption) {
 	fs.grpcDialOption = option
 }
 
+func (fs *FilerSource) SetHttpClient(client *util_http_client.HTTPClient) {
+	fs.httpClient = client
+}
+
 func (fs *FilerSource) LookupFileId(ctx context.Context, part string) (fileUrls []string, err error) {
 
 	vid2Locations := make(map[string]*filer_pb.Locations)
@@ -104,9 +110,15 @@ func (fs *FilerSource) LookupFileId(ctx context.Context, part string) (fileUrls
 }
 
 func (fs *FilerSource) ReadPart(fileId string, offset int64) (filename string, header http.Header, resp *http.Response, err error) {
+	downloadFn := util_http.DownloadFile
+	if fs.httpClient != nil {
+		downloadFn = func(fileUrl string, jwt string, offset ...int64) (string, http.Header, *http.Response, error) {
+			return util_http.DownloadFileWithClient(fs.httpClient, fileUrl, jwt, offset...)
+		}
+	}
 
 	if fs.proxyByFiler {
-		filename, header, resp, err = util_http.DownloadFile("http://"+fs.address+"/?proxyChunkId="+fileId, "", offset)
+		filename, header, resp, err = downloadFn("http://"+fs.address+"/?proxyChunkId="+fileId, "", offset)
 		if err != nil {
 			glog.V(0).Infof("read part %s via filer proxy %s offset %d: %v", fileId, fs.address, offset, err)
 		} else {
@@ -121,7 +133,7 @@ func (fs *FilerSource) ReadPart(fileId string, offset int64) (filename string, h
 	}
 
 	for _, fileUrl := range fileUrls {
-		filename, header, resp, err = util_http.DownloadFile(fileUrl, "", offset)
+		filename, header, resp, err = downloadFn(fileUrl, "", offset)
 		if err != nil {
 			glog.V(0).Infof("fail to read part %s from %s offset %d: %v", fileId, fileUrl, offset, err)
 		} else {

weed/s3api/auth_signature_v4.go

@@ -194,6 +194,8 @@ func buildPathWithForwardedPrefix(forwardedPrefix, urlPath string) string {
 	var joined string
 	if strings.HasSuffix(forwardedPrefix, "/") && strings.HasPrefix(urlPath, "/") {
 		joined = forwardedPrefix + urlPath[1:]
+	} else if urlPath == "" {
+		joined = forwardedPrefix
 	} else if !strings.HasSuffix(forwardedPrefix, "/") && !strings.HasPrefix(urlPath, "/") {
 		joined = forwardedPrefix + "/" + urlPath
 	} else {

weed/s3api/auth_signature_v4_test.go

@@ -139,6 +139,12 @@ func TestBuildPathWithForwardedPrefix(t *testing.T) {
 			urlPath:         "bucket/obj",
 			expected:        "/storage/bucket/obj",
 		},
+		{
+			name:            "empty urlPath with prefix",
+			forwardedPrefix: "/s3",
+			urlPath:         "",
+			expected:        "/s3",
+		},
 	}
 
 	for _, tt := range tests {

weed/s3api/iceberg/commit_helpers.go

@@ -200,7 +200,7 @@ func (s *Server) finalizeCreateOnCommit(ctx context.Context, input createOnCommi
 		markerBucket = metadataBucket
 	}
 	if markerErr := s.deleteStageCreateMarkers(ctx, markerBucket, input.namespace, input.tableName); markerErr != nil {
-		glog.V(1).Infof("Iceberg: failed to cleanup stage-create markers for %s.%s after finalize: %v", encodeNamespace(input.namespace), input.tableName, markerErr)
+		glog.V(1).Infof("Iceberg: failed to cleanup stage-create markers for %s.%s after finalize: %v", flattenNamespacePath(input.namespace), input.tableName, markerErr)
 	}
 
 	return &CommitTableResponse{

weed/s3api/iceberg/handlers_commit.go

@@ -83,7 +83,7 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) {
 		})
 		if err != nil {
 			if isS3TablesNotFound(err) {
-				location := fmt.Sprintf("s3://%s/%s/%s", bucketName, encodeNamespace(namespace), tableName)
+				location := fmt.Sprintf("s3://%s/%s", bucketName, path.Join(flattenNamespacePath(namespace), tableName))
 				tableUUID := generatedLegacyUUID
 				baseMetadataVersion := 0
 				baseMetadataLocation := ""
@@ -195,7 +195,7 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) {
 
 		location := tableLocationFromMetadataLocation(getResp.MetadataLocation)
 		if location == "" {
-			location = fmt.Sprintf("s3://%s/%s/%s", bucketName, encodeNamespace(namespace), tableName)
+			location = fmt.Sprintf("s3://%s/%s", bucketName, path.Join(flattenNamespacePath(namespace), tableName))
 		}
 		tableUUID := uuid.Nil
 		if getResp.Metadata != nil && getResp.Metadata.Iceberg != nil && getResp.Metadata.Iceberg.TableUUID != "" {

weed/s3api/iceberg/handlers_namespace.go

@@ -39,10 +39,21 @@ func (s *Server) handleListNamespaces(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// The Iceberg REST spec allows a "parent" query parameter for hierarchical
+	// namespace listing. Convert it to the dot-separated prefix used by S3 Tables.
+	var prefix string
+	if parent := r.URL.Query().Get("parent"); parent != "" {
+		parentParts := parseNamespace(parent)
+		if len(parentParts) > 0 {
+			prefix = flattenNamespacePath(parentParts) + "."
+		}
+	}
+
 	// Use S3 Tables manager to list namespaces
 	var resp s3tables.ListNamespacesResponse
 	req := &s3tables.ListNamespacesRequest{
 		TableBucketARN:    bucketARN,
+		Prefix:            prefix,
 		ContinuationToken: pageToken,
 		MaxNamespaces:     pageSize,
 	}

weed/s3api/iceberg/handlers_table.go

@@ -120,7 +120,7 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
 
 	// Generate UUID for the new table
 	tableUUID := uuid.New()
-	tablePath := path.Join(encodeNamespace(namespace), req.Name)
+	tablePath := path.Join(flattenNamespacePath(namespace), req.Name)
 	location := strings.TrimSuffix(req.Location, "/")
 	if location == "" {
 		if req.Properties != nil {
@@ -179,7 +179,7 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
 		}
 		stagedMetadataLocation := fmt.Sprintf("s3://%s/%s/metadata/%s", metadataBucket, stagedTablePath, metadataFileName)
 		if markerErr := s.writeStageCreateMarker(r.Context(), bucketName, namespace, tableName, tableUUID, location, stagedMetadataLocation); markerErr != nil {
-			glog.V(1).Infof("Iceberg: failed to persist stage-create marker for %s.%s: %v", encodeNamespace(namespace), tableName, markerErr)
+			glog.V(1).Infof("Iceberg: failed to persist stage-create marker for %s.%s: %v", flattenNamespacePath(namespace), tableName, markerErr)
 		}
 		result := LoadTableResult{
 			MetadataLocation: metadataLocation,
@@ -266,7 +266,7 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
 		finalLocation = metadataLocation
 	}
 	if markerErr := s.deleteStageCreateMarkers(r.Context(), bucketName, namespace, tableName); markerErr != nil {
-		glog.V(1).Infof("Iceberg: failed to cleanup stage-create markers for %s.%s after create: %v", encodeNamespace(namespace), tableName, markerErr)
+		glog.V(1).Infof("Iceberg: failed to cleanup stage-create markers for %s.%s after create: %v", flattenNamespacePath(namespace), tableName, markerErr)
 	}
 
 	result := LoadTableResult{
@@ -324,7 +324,7 @@ func (s *Server) handleLoadTable(w http.ResponseWriter, r *http.Request) {
 func buildLoadTableResult(getResp s3tables.GetTableResponse, bucketName string, namespace []string, tableName string) LoadTableResult {
 	location := tableLocationFromMetadataLocation(getResp.MetadataLocation)
 	if location == "" {
-		location = fmt.Sprintf("s3://%s/%s/%s", bucketName, encodeNamespace(namespace), tableName)
+		location = fmt.Sprintf("s3://%s/%s", bucketName, path.Join(flattenNamespacePath(namespace), tableName))
 	}
 	tableUUID := uuid.Nil
 	if getResp.Metadata != nil && getResp.Metadata.Iceberg != nil && getResp.Metadata.Iceberg.TableUUID != "" {

weed/s3api/iceberg/utils.go

@@ -32,11 +32,19 @@ func parseNamespace(encoded string) []string {
 	return result
 }
 
-// encodeNamespace encodes namespace parts for response.
+// encodeNamespace encodes namespace parts using the Iceberg REST protocol's
+// unit separator (0x1F) convention. This is only appropriate for protocol-level
+// encoding (e.g. URL path parameters), NOT for filesystem/S3 paths.
 func encodeNamespace(parts []string) string {
 	return strings.Join(parts, "\x1F")
 }
 
+// flattenNamespacePath joins namespace parts with "." for use in S3 location
+// and filer paths, matching the S3 Tables storage layer convention.
+func flattenNamespacePath(parts []string) string {
+	return strings.Join(parts, ".")
+}
+
 func parseS3Location(location string) (bucketName, tablePath string, err error) {
 	if !strings.HasPrefix(location, "s3://") {
 		return "", "", fmt.Errorf("unsupported location: %s", location)

weed/s3api/policy_engine/engine.go

@@ -162,6 +162,16 @@ func (engine *PolicyEngine) evaluateStatement(stmt *CompiledStatement, args *Pol
 	if !matchedAction {
 		matchedAction = engine.matchesDynamicPatterns(stmt.DynamicActionPatterns, args.Action, args)
 	}
+	// Multipart upload actions (CreateMultipartUpload, UploadPart, CompleteMultipartUpload, etc.)
+	// are implicitly allowed by s3:PutObject, since multipart upload is an implementation
+	// detail of putting objects. Check if this is a multipart action and the statement
+	// grants s3:PutObject.
+	if !matchedAction && multipartActionSet[args.Action] {
+		matchedAction = engine.matchesPatterns(stmt.ActionPatterns, "s3:PutObject")
+		if !matchedAction {
+			matchedAction = engine.matchesDynamicPatterns(stmt.DynamicActionPatterns, "s3:PutObject", args)
+		}
+	}
 	if !matchedAction {
 		return false
 	}

weed/s3api/policy_engine/engine_test.go

@@ -981,3 +981,80 @@ func TestExistingObjectTagDenyPolicy(t *testing.T) {
 		})
 	}
 }
+
+// TestMultipartUploadInheritsPutObjectPermission verifies that granting s3:PutObject
+// in a bucket policy implicitly allows multipart upload operations.
+// See https://github.com/seaweedfs/seaweedfs/discussions/8751
+func TestMultipartUploadInheritsPutObjectPermission(t *testing.T) {
+	engine := NewPolicyEngine()
+
+	policyJSON := `{
+		"Version": "2012-10-17",
+		"Statement": [
+			{
+				"Effect": "Allow",
+				"Principal": "*",
+				"Action": "s3:PutObject",
+				"Resource": "arn:aws:s3:::test-bucket/*"
+			}
+		]
+	}`
+
+	if err := engine.SetBucketPolicy("test-bucket", policyJSON); err != nil {
+		t.Fatalf("Failed to set policy: %v", err)
+	}
+
+	multipartActions := []string{
+		"s3:CreateMultipartUpload",
+		"s3:UploadPart",
+		"s3:UploadPartCopy",
+		"s3:CompleteMultipartUpload",
+		"s3:AbortMultipartUpload",
+		"s3:ListMultipartUploadParts",
+		"s3:ListBucketMultipartUploads",
+	}
+
+	for _, action := range multipartActions {
+		t.Run(action, func(t *testing.T) {
+			args := &PolicyEvaluationArgs{
+				Action:    action,
+				Resource:  "arn:aws:s3:::test-bucket/myfile.dat",
+				Principal: "*",
+				Conditions: map[string][]string{
+					"aws:SourceIp": {"10.0.0.1"},
+				},
+			}
+			result := engine.EvaluatePolicy("test-bucket", args)
+			if result != PolicyResultAllow {
+				t.Errorf("Expected s3:PutObject to implicitly allow %s, got %v", action, result)
+			}
+		})
+	}
+
+	// ListBucketMultipartUploads is a bucket-level action; the object-only
+	// resource "arn:aws:s3:::test-bucket/*" should NOT match the bucket ARN.
+	t.Run("s3:ListBucketMultipartUploads bucket ARN", func(t *testing.T) {
+		args := &PolicyEvaluationArgs{
+			Action:    "s3:ListBucketMultipartUploads",
+			Resource:  "arn:aws:s3:::test-bucket",
+			Principal: "*",
+		}
+		result := engine.EvaluatePolicy("test-bucket", args)
+		if result == PolicyResultAllow {
+			t.Error("Object-only resource should not match bucket ARN for ListBucketMultipartUploads")
+		}
+	})
+
+	// s3:PutObject must NOT implicitly grant unrelated actions
+	t.Run("s3:DeleteObject not inherited", func(t *testing.T) {
+		args := &PolicyEvaluationArgs{
+			Action:    "s3:DeleteObject",
+			Resource:  "arn:aws:s3:::test-bucket/myfile.dat",
+			Principal: "*",
+		}
+		result := engine.EvaluatePolicy("test-bucket", args)
+		if result == PolicyResultAllow {
+			t.Error("s3:PutObject should NOT implicitly allow s3:DeleteObject")
+		}
+	})
+}

weed/s3api/policy_engine/types.go

@@ -43,6 +43,7 @@ var (
 	multipartActionSet = map[string]bool{
 		s3const.S3_ACTION_CREATE_MULTIPART:       true,
 		s3const.S3_ACTION_UPLOAD_PART:            true,
+		s3const.S3_ACTION_UPLOAD_PART_COPY:       true,
 		s3const.S3_ACTION_COMPLETE_MULTIPART:     true,
 		s3const.S3_ACTION_ABORT_MULTIPART:        true,
 		s3const.S3_ACTION_LIST_PARTS:             true,

weed/s3api/s3_constants/s3_action_strings.go

@@ -32,6 +32,7 @@ const (
 	S3_ACTION_UPLOAD_PART            = "s3:UploadPart"
 	S3_ACTION_COMPLETE_MULTIPART     = "s3:CompleteMultipartUpload"
 	S3_ACTION_ABORT_MULTIPART        = "s3:AbortMultipartUpload"
+	S3_ACTION_UPLOAD_PART_COPY       = "s3:UploadPartCopy"
 	S3_ACTION_LIST_PARTS             = "s3:ListMultipartUploadParts"
 	S3_ACTION_LIST_MULTIPART_UPLOADS = "s3:ListBucketMultipartUploads"
 

weed/s3api/s3_end_to_end_test.go

@@ -214,6 +214,153 @@ func TestS3MultipartUploadWithJWT(t *testing.T) {
 	}
 }
 
+// TestS3ListObjectsV2PrefixCondition tests that ListObjectsV2 requests with a prefix
+// query parameter correctly populate s3:prefix in the policy evaluation context and
+// use bucket-level resource ARNs, so that policies with s3:prefix conditions work.
+// This reproduces the bug reported in https://github.com/seaweedfs/seaweedfs/issues/8969
+func TestS3ListObjectsV2PrefixCondition(t *testing.T) {
+	// Set up IAM system
+	iamManager := integration.NewIAMManager()
+	config := &integration.IAMConfig{
+		STS: &sts.STSConfig{
+			TokenDuration:    sts.FlexibleDuration{Duration: time.Hour},
+			MaxSessionLength: sts.FlexibleDuration{Duration: time.Hour * 12},
+			Issuer:           "test-sts",
+			SigningKey:       []byte("test-signing-key-32-characters-long"),
+		},
+		Policy: &policy.PolicyEngineConfig{
+			DefaultEffect: "Deny",
+			StoreType:     "memory",
+		},
+		Roles: &integration.RoleStoreConfig{
+			StoreType: "memory",
+		},
+	}
+
+	err := iamManager.Initialize(config, func() string { return "localhost:8888" })
+	require.NoError(t, err)
+
+	setupTestProviders(t, iamManager)
+
+	s3IAMIntegration := NewS3IAMIntegration(iamManager, "localhost:8888")
+	require.NotNil(t, s3IAMIntegration)
+
+	ctx := context.Background()
+
+	// Create a role with a policy that allows ListBucket only with a specific s3:prefix condition.
+	// This is the pattern used by Lakekeeper-vended STS credentials (issue #8969).
+	prefixPolicy := &policy.PolicyDocument{
+		Version: "2012-10-17",
+		Statement: []policy.Statement{
+			{
+				Sid:    "AllowListUnderWarehouse",
+				Effect: "Allow",
+				Action: []string{"s3:ListBucket"},
+				Resource: []string{
+					"arn:aws:s3:::examples",
+				},
+				Condition: map[string]map[string]interface{}{
+					"StringLike": {
+						"s3:prefix": []string{"warehouse/*"},
+					},
+				},
+			},
+			{
+				Sid:      "AllowSTSSessionValidation",
+				Effect:   "Allow",
+				Action:   []string{"sts:ValidateSession"},
+				Resource: []string{"*"},
+			},
+		},
+	}
+
+	iamManager.CreatePolicy(ctx, "", "PrefixRestrictedPolicy", prefixPolicy)
+	iamManager.CreateRole(ctx, "", "PrefixRestrictedRole", &integration.RoleDefinition{
+		RoleName: "PrefixRestrictedRole",
+		TrustPolicy: &policy.PolicyDocument{
+			Version: "2012-10-17",
+			Statement: []policy.Statement{
+				{
+					Effect:    "Allow",
+					Principal: map[string]interface{}{"Federated": "test-oidc"},
+					Action:    []string{"sts:AssumeRoleWithWebIdentity"},
+				},
+			},
+		},
+		AttachedPolicies: []string{"PrefixRestrictedPolicy"},
+	})
+
+	// Assume role to get a session token
+	validJWTToken := createTestJWTEndToEnd(t, "https://test-issuer.com", "test-user-123", "test-signing-key")
+	response, err := iamManager.AssumeRoleWithWebIdentity(ctx, &sts.AssumeRoleWithWebIdentityRequest{
+		RoleArn:          "arn:aws:iam::role/PrefixRestrictedRole",
+		WebIdentityToken: validJWTToken,
+		RoleSessionName:  "prefix-test-session",
+	})
+	require.NoError(t, err)
+
+	sessionToken := response.Credentials.SessionToken
+	require.NotEmpty(t, sessionToken)
+
+	// Authenticate to get IAM identity
+	authReq := httptest.NewRequest("GET", "/examples", http.NoBody)
+	authReq.Header.Set("Authorization", "Bearer "+sessionToken)
+	identity, errCode := s3IAMIntegration.AuthenticateJWT(ctx, authReq)
+	require.Equal(t, s3err.ErrNone, errCode, "Authentication should succeed")
+
+	tests := []struct {
+		name     string
+		url      string
+		bucket   string
+		objKey   string
+		expected s3err.ErrorCode
+	}{
+		{
+			name:     "ListObjectsV2 with matching prefix query param and empty objectKey",
+			url:      "/examples?list-type=2&prefix=warehouse/data",
+			bucket:   "examples",
+			objKey:   "",
+			expected: s3err.ErrNone,
+		},
+		{
+			name:     "ListObjectsV2 with matching prefix propagated as objectKey",
+			url:      "/examples?list-type=2&prefix=warehouse/data",
+			bucket:   "examples",
+			objKey:   "warehouse/data",
+			expected: s3err.ErrNone,
+		},
+		{
+			name:     "ListObjectsV1 with matching prefix query param",
+			url:      "/examples?prefix=warehouse/files",
+			bucket:   "examples",
+			objKey:   "",
+			expected: s3err.ErrNone,
+		},
+		{
+			name:     "ListObjectsV2 with non-matching prefix should be denied",
+			url:      "/examples?list-type=2&prefix=other/path",
+			bucket:   "examples",
+			objKey:   "",
+			expected: s3err.ErrAccessDenied,
+		},
+		{
+			name:     "ListObjectsV2 with no prefix should be denied",
+			url:      "/examples?list-type=2",
+			bucket:   "examples",
+			objKey:   "",
+			expected: s3err.ErrAccessDenied,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			req := httptest.NewRequest("GET", tt.url, http.NoBody)
+			result := s3IAMIntegration.AuthorizeAction(ctx, identity, Action("List"), tt.bucket, tt.objKey, req)
+			assert.Equal(t, tt.expected, result, "unexpected authorization result for %s", tt.name)
+		})
+	}
+}
+
 // TestS3CORSWithJWT tests CORS preflight requests with IAM
 func TestS3CORSWithJWT(t *testing.T) {
 	s3Server, iamManager := setupCompleteS3IAMSystem(t)

weed/s3api/s3_iam_middleware.go

@@ -250,20 +250,28 @@ func (s3iam *S3IAMIntegration) AuthorizeAction(ctx context.Context, identity *IA
 		return s3err.ErrAccessDenied
 	}
 
-	// Build resource ARN for the S3 operation
-	resourceArn := buildS3ResourceArn(bucket, objectKey)
-
 	// Extract request context for policy conditions
 	requestContext := extractRequestContext(r)
 
-	// Add s3:prefix to request context based on object key
-	// This ensures that policy conditions referencing s3:prefix (like StringLike)
-	// work correctly for both ListObjects (where objectKey is the prefix) and
-	// object operations (where we treat the object key as the prefix for matching)
-	if objectKey != "" && objectKey != "/" {
-		requestContext["s3:prefix"] = objectKey
+	// For list operations, populate the s3:prefix condition key and ensure the
+	// resource ARN stays at bucket level (matching AWS ListBucket semantics).
+	// See https://github.com/seaweedfs/seaweedfs/issues/8969
+	resourceObjectKey := objectKey
+	if action == "List" {
+		listPrefix := r.URL.Query().Get("prefix")
+		if listPrefix != "" {
+			requestContext["s3:prefix"] = listPrefix
+		} else if objectKey != "" && objectKey != "/" {
+			requestContext["s3:prefix"] = objectKey
+		} else {
+			requestContext["s3:prefix"] = ""
+		}
+		resourceObjectKey = ""
 	}
 
+	// Build resource ARN for the S3 operation
+	resourceArn := buildS3ResourceArn(bucket, resourceObjectKey)
+
 	// Add identity claims to request context for policy variables
 	// Only add claim keys if they don't already exist (to avoid overwriting request-derived context)
 	if identity.Claims != nil {

weed/security/tls.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/util"
+	util_http_client "github.com/seaweedfs/seaweedfs/weed/util/http/client"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
@@ -209,6 +210,39 @@ func LoadClientTLS(config *util.ViperProxy, component string) grpc.DialOption {
 	return grpc.WithTransportCredentials(wrapped)
 }
 
+// LoadHTTPClientFromFile creates an HTTP client using the https.client TLS
+// settings from the given security config file. Returns nil if HTTPS is not
+// enabled in the config. This is used by filer.sync to create per-cluster
+// HTTP clients when clusters use different certificates.
+func LoadHTTPClientFromFile(configFile string) (*util_http_client.HTTPClient, error) {
+	v := viper.New()
+	v.SetConfigFile(configFile)
+	if err := v.ReadInConfig(); err != nil {
+		return nil, fmt.Errorf("failed to read security config %s: %v", configFile, err)
+	}
+
+	if !v.GetBool("https.client.enabled") {
+		return nil, nil
+	}
+
+	configDir := filepath.Dir(configFile)
+	resolvePath := func(key string) string {
+		p := v.GetString(key)
+		if p != "" && !filepath.IsAbs(p) {
+			return filepath.Join(configDir, p)
+		}
+		return p
+	}
+
+	return util_http_client.NewHttpClientWithTLS(
+		resolvePath("https.client.cert"),
+		resolvePath("https.client.key"),
+		resolvePath("https.client.ca"),
+		v.GetBool("https.client.insecure_skip_verify"),
+		util_http_client.AddDialContext,
+	)
+}
+
 func LoadClientTLSHTTP(clientCertFile string) *tls.Config {
 	clientCerts, err := os.ReadFile(clientCertFile)
 	if err != nil {

weed/server/filer_server.go

@@ -248,6 +248,10 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 			glog.Fatalf("%s bootstrap from %+v: %v", option.Host, existingNodes, err)
 		}
 	}
+	v.SetDefault("filer.options.s3.empty_folder_cleanup_delay", "2m")
+	if d, err := time.ParseDuration(v.GetString("filer.options.s3.empty_folder_cleanup_delay")); err == nil {
+		fs.filer.EmptyFolderCleanupDelay = d
+	}
 	fs.filer.AggregateFromPeers(option.Host, existingNodes, startFromTime)
 
 	fs.filer.LoadFilerConf()

weed/server/volume_grpc_erasure_coding.go

@@ -86,16 +86,30 @@ func (vs *VolumeServer) VolumeEcShardsGenerate(ctx context.Context, req *volume_
 		os.Remove(v.IndexFileName() + ".ecx")
 	}()
 
+	// IMPORTANT: Generate .ecx BEFORE EC shards to prevent a race condition.
+	// If .ecx were generated after EC shards, any write (e.g. from WriteNeedleBlob
+	// during replica sync) between the two steps would add entries to .idx that
+	// end up in .ecx but whose data is NOT in the EC shards — causing "shard too
+	// short" and "size mismatch" errors on reads.
+	//
+	// By generating .ecx first, it reflects the .idx state at or before the .dat
+	// is read for EC encoding. If a write sneaks in after .ecx but before/during
+	// EC encoding, the shards contain MORE data than .ecx references, which is
+	// harmless (the extra data is simply not indexed).
+
+	// write .ecx file from the current .idx
+	if err := erasure_coding.WriteSortedFileFromIdx(v.IndexFileName(), ".ecx"); err != nil {
+		return nil, fmt.Errorf("WriteSortedFileFromIdx %s: %v", v.IndexFileName(), err)
+	}
+
+	// snapshot .dat file size before encoding — must match what .ecx references
+	datSize, _, _ := v.FileStat()
+
 	// write .ec00 ~ .ec[TotalShards-1] files using context
 	if err := erasure_coding.WriteEcFilesWithContext(baseFileName, ecCtx); err != nil {
 		return nil, fmt.Errorf("WriteEcFilesWithContext %s: %v", baseFileName, err)
 	}
 
-	// write .ecx file
-	if err := erasure_coding.WriteSortedFileFromIdx(v.IndexFileName(), ".ecx"); err != nil {
-		return nil, fmt.Errorf("WriteSortedFileFromIdx %s: %v", v.IndexFileName(), err)
-	}
-
 	// write .vif files
 	var expireAtSec uint64
 	if v.Ttl != nil {
@@ -106,8 +120,6 @@ func (vs *VolumeServer) VolumeEcShardsGenerate(ctx context.Context, req *volume_
 	}
 	volumeInfo := &volume_server_pb.VolumeInfo{Version: uint32(v.Version())}
 	volumeInfo.ExpireAtSec = expireAtSec
-
-	datSize, _, _ := v.FileStat()
 	volumeInfo.DatFileSize = int64(datSize)
 
 	// Validate EC configuration before saving to .vif

weed/shell/command.go

@@ -17,4 +17,5 @@ type CommandTag string
 
 const (
 	ResourceHeavy CommandTag = "resourceHeavy"
+	Hidden        CommandTag = "hidden"
 )

weed/shell/command_s3_accesskey_create.go

@@ -0,0 +1,96 @@
+package shell
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/iam"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3AccessKeyCreate{})
+}
+
+type commandS3AccessKeyCreate struct {
+}
+
+func (c *commandS3AccessKeyCreate) Name() string {
+	return "s3.accesskey.create"
+}
+
+func (c *commandS3AccessKeyCreate) Help() string {
+	return `create an additional access key for an S3 IAM user
+
+	s3.accesskey.create -user <username>
+	s3.accesskey.create -user <username> -access_key <key> -secret_key <secret>
+
+	Generates a new credential pair for an existing user. If -access_key and
+	-secret_key are omitted, they are generated automatically.
+`
+}
+
+func (c *commandS3AccessKeyCreate) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3AccessKeyCreate) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	user := f.String("user", "", "user name")
+	accessKey := f.String("access_key", "", "access key (generated if omitted)")
+	secretKey := f.String("secret_key", "", "secret key (generated if omitted)")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	if *user == "" {
+		return fmt.Errorf("-user is required")
+	}
+
+	ak := *accessKey
+	sk := *secretKey
+
+	if ak == "" && sk == "" {
+		var err error
+		ak, err = iam.GenerateRandomString(iam.AccessKeyIdLength, iam.CharsetUpper)
+		if err != nil {
+			return fmt.Errorf("generate access key: %v", err)
+		}
+		sk, err = iam.GenerateSecretAccessKey()
+		if err != nil {
+			return fmt.Errorf("generate secret key: %v", err)
+		}
+	} else if ak == "" || sk == "" {
+		return fmt.Errorf("both -access_key and -secret_key must be provided together, or omit both to auto-generate")
+	}
+
+	err := pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+		_, err := client.CreateAccessKey(ctx, &iam_pb.CreateAccessKeyRequest{
+			Username: *user,
+			Credential: &iam_pb.Credential{
+				AccessKey: ak,
+				SecretKey: sk,
+				Status:    iam.AccessKeyStatusActive,
+			},
+		})
+		return err
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+	if err != nil {
+		return err
+	}
+
+	fmt.Fprintf(writer, "Created access key for user %q\n", *user)
+	fmt.Fprintf(writer, "Access Key: %s\n", ak)
+	fmt.Fprintf(writer, "Secret Key: %s\n", sk)
+	fmt.Fprintln(writer)
+	fmt.Fprintln(writer, "Save these credentials - the secret key cannot be retrieved later.")
+	return nil
+}

weed/shell/command_s3_accesskey_delete.go

@@ -0,0 +1,68 @@
+package shell
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3AccessKeyDelete{})
+}
+
+type commandS3AccessKeyDelete struct {
+}
+
+func (c *commandS3AccessKeyDelete) Name() string {
+	return "s3.accesskey.delete"
+}
+
+func (c *commandS3AccessKeyDelete) Help() string {
+	return `delete an access key from an S3 IAM user
+
+	s3.accesskey.delete -user <username> -access_key <key>
+`
+}
+
+func (c *commandS3AccessKeyDelete) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3AccessKeyDelete) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	user := f.String("user", "", "user name")
+	accessKey := f.String("access_key", "", "access key to delete")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	if *user == "" {
+		return fmt.Errorf("-user is required")
+	}
+	if *accessKey == "" {
+		return fmt.Errorf("-access_key is required")
+	}
+
+	err := pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+		_, err := client.DeleteAccessKey(ctx, &iam_pb.DeleteAccessKeyRequest{
+			Username:  *user,
+			AccessKey: *accessKey,
+		})
+		return err
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+	if err != nil {
+		return err
+	}
+
+	fmt.Fprintf(writer, "Deleted access key %s from user %q\n", *accessKey, *user)
+	return nil
+}

weed/shell/command_s3_accesskey_list.go

@@ -0,0 +1,75 @@
+package shell
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"text/tabwriter"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3AccessKeyList{})
+}
+
+type commandS3AccessKeyList struct {
+}
+
+func (c *commandS3AccessKeyList) Name() string {
+	return "s3.accesskey.list"
+}
+
+func (c *commandS3AccessKeyList) Help() string {
+	return `list access keys for an S3 IAM user
+
+	s3.accesskey.list -user <username>
+`
+}
+
+func (c *commandS3AccessKeyList) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3AccessKeyList) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	user := f.String("user", "", "user name")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	if *user == "" {
+		return fmt.Errorf("-user is required")
+	}
+
+	return pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		resp, err := client.GetUser(ctx, &iam_pb.GetUserRequest{Username: *user})
+		if err != nil {
+			return err
+		}
+
+		if len(resp.Identity.Credentials) == 0 {
+			fmt.Fprintf(writer, "No access keys for user %q.\n", *user)
+			return nil
+		}
+
+		tw := tabwriter.NewWriter(writer, 0, 4, 2, ' ', 0)
+		fmt.Fprintln(tw, "ACCESS KEY\tSTATUS")
+		for _, cred := range resp.Identity.Credentials {
+			st := cred.Status
+			if st == "" {
+				st = "Active"
+			}
+			fmt.Fprintf(tw, "%s\t%s\n", cred.AccessKey, st)
+		}
+		return tw.Flush()
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+}

weed/shell/command_s3_accesskey_rotate.go

@@ -0,0 +1,105 @@
+package shell
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/iam"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3AccessKeyRotate{})
+}
+
+type commandS3AccessKeyRotate struct {
+}
+
+func (c *commandS3AccessKeyRotate) Name() string {
+	return "s3.accesskey.rotate"
+}
+
+func (c *commandS3AccessKeyRotate) Help() string {
+	return `rotate an access key for an S3 IAM user
+
+	s3.accesskey.rotate -user <username> -access_key <old_key>
+
+	Creates a new credential pair and deletes the old one. There is a brief
+	window where both keys are valid.
+`
+}
+
+func (c *commandS3AccessKeyRotate) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3AccessKeyRotate) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	user := f.String("user", "", "user name")
+	oldKey := f.String("access_key", "", "access key to rotate")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	if *user == "" {
+		return fmt.Errorf("-user is required")
+	}
+	if *oldKey == "" {
+		return fmt.Errorf("-access_key is required")
+	}
+
+	newAK, err := iam.GenerateRandomString(iam.AccessKeyIdLength, iam.CharsetUpper)
+	if err != nil {
+		return fmt.Errorf("generate access key: %v", err)
+	}
+	newSK, err := iam.GenerateSecretAccessKey()
+	if err != nil {
+		return fmt.Errorf("generate secret key: %v", err)
+	}
+
+	err = pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		// Create new key first so there's no gap without credentials
+		_, err := client.CreateAccessKey(ctx, &iam_pb.CreateAccessKeyRequest{
+			Username: *user,
+			Credential: &iam_pb.Credential{
+				AccessKey: newAK,
+				SecretKey: newSK,
+				Status:    iam.AccessKeyStatusActive,
+			},
+		})
+		if err != nil {
+			return fmt.Errorf("create new key: %v", err)
+		}
+
+		// Delete old key
+		_, err = client.DeleteAccessKey(ctx, &iam_pb.DeleteAccessKeyRequest{
+			Username:  *user,
+			AccessKey: *oldKey,
+		})
+		if err != nil {
+			return fmt.Errorf("delete old key (new key %s was already created): %v", newAK, err)
+		}
+
+		return nil
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+	if err != nil {
+		return err
+	}
+
+	fmt.Fprintf(writer, "Rotated access key for user %q\n", *user)
+	fmt.Fprintf(writer, "Old Key:    %s (deleted)\n", *oldKey)
+	fmt.Fprintf(writer, "Access Key: %s\n", newAK)
+	fmt.Fprintf(writer, "Secret Key: %s\n", newSK)
+	fmt.Fprintln(writer)
+	fmt.Fprintln(writer, "Save these credentials - the secret key cannot be retrieved later.")
+	return nil
+}

weed/shell/command_s3_anonymous_get.go

@@ -0,0 +1,89 @@
+package shell
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3AnonymousGet{})
+}
+
+type commandS3AnonymousGet struct {
+}
+
+func (c *commandS3AnonymousGet) Name() string {
+	return "s3.anonymous.get"
+}
+
+func (c *commandS3AnonymousGet) Help() string {
+	return `show anonymous access for a bucket
+
+	s3.anonymous.get -bucket <bucket_name>
+`
+}
+
+func (c *commandS3AnonymousGet) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3AnonymousGet) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	bucket := f.String("bucket", "", "bucket name")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	if *bucket == "" {
+		return fmt.Errorf("-bucket is required")
+	}
+
+	return pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		resp, err := client.GetUser(ctx, &iam_pb.GetUserRequest{Username: anonymousUserName})
+		if err != nil {
+			st, ok := status.FromError(err)
+			if ok && st.Code() == codes.NotFound {
+				fmt.Fprintf(writer, "Bucket: %s\nAccess: none\n", *bucket)
+				return nil
+			}
+			return err
+		}
+		if resp.Identity == nil {
+			fmt.Fprintf(writer, "Bucket: %s\nAccess: none\n", *bucket)
+			return nil
+		}
+
+		var actions []string
+		for _, a := range resp.Identity.Actions {
+			parts := strings.SplitN(a, ":", 2)
+			if len(parts) == 2 && parts[1] == *bucket {
+				actions = append(actions, parts[0])
+			}
+		}
+
+		fmt.Fprintf(writer, "Bucket: %s\n", *bucket)
+		if len(actions) == 0 {
+			fmt.Fprintln(writer, "Access: none")
+		} else {
+			sort.Strings(actions)
+			fmt.Fprintf(writer, "Access: %s\n", strings.Join(actions, ", "))
+		}
+
+		return nil
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+}

weed/shell/command_s3_anonymous_list.go

@@ -0,0 +1,91 @@
+package shell
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"sort"
+	"strings"
+	"text/tabwriter"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3AnonymousList{})
+}
+
+type commandS3AnonymousList struct {
+}
+
+func (c *commandS3AnonymousList) Name() string {
+	return "s3.anonymous.list"
+}
+
+func (c *commandS3AnonymousList) Help() string {
+	return `list all buckets with anonymous access
+
+	s3.anonymous.list
+`
+}
+
+func (c *commandS3AnonymousList) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3AnonymousList) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	return pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		resp, err := client.GetUser(ctx, &iam_pb.GetUserRequest{Username: anonymousUserName})
+		if err != nil {
+			st, ok := status.FromError(err)
+			if ok && st.Code() == codes.NotFound {
+				fmt.Fprintln(writer, "No anonymous access configured.")
+				return nil
+			}
+			return err
+		}
+		if resp.Identity == nil {
+			fmt.Fprintln(writer, "No anonymous access configured.")
+			return nil
+		}
+
+		// Group actions by bucket
+		bucketActions := map[string][]string{}
+		for _, a := range resp.Identity.Actions {
+			parts := strings.SplitN(a, ":", 2)
+			if len(parts) == 2 {
+				bucketActions[parts[1]] = append(bucketActions[parts[1]], parts[0])
+			}
+		}
+
+		if len(bucketActions) == 0 {
+			fmt.Fprintln(writer, "No anonymous access configured.")
+			return nil
+		}
+
+		// Sort bucket names
+		buckets := make([]string, 0, len(bucketActions))
+		for b := range bucketActions {
+			buckets = append(buckets, b)
+		}
+		sort.Strings(buckets)
+
+		tw := tabwriter.NewWriter(writer, 0, 4, 2, ' ', 0)
+		fmt.Fprintln(tw, "BUCKET\tACCESS")
+		for _, b := range buckets {
+			actions := bucketActions[b]
+			sort.Strings(actions)
+			fmt.Fprintf(tw, "%s\t%s\n", b, strings.Join(actions, ", "))
+		}
+		return tw.Flush()
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+}

weed/shell/command_s3_anonymous_set.go

@@ -0,0 +1,144 @@
+package shell
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"strings"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+const anonymousUserName = "anonymous"
+
+func init() {
+	Commands = append(Commands, &commandS3AnonymousSet{})
+}
+
+type commandS3AnonymousSet struct {
+}
+
+func (c *commandS3AnonymousSet) Name() string {
+	return "s3.anonymous.set"
+}
+
+func (c *commandS3AnonymousSet) Help() string {
+	return `set anonymous (public) access on a bucket
+
+	s3.anonymous.set -bucket <bucket_name> -access Read,List
+	s3.anonymous.set -bucket <bucket_name> -access none
+
+	Supported actions: Read, Write, List, Tagging, Admin
+	Use "none" to remove all anonymous access for the bucket.
+
+	This manages the special "anonymous" user's actions. It does not
+	use IAM policies — it sets legacy per-bucket actions directly.
+`
+}
+
+func (c *commandS3AnonymousSet) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3AnonymousSet) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	bucket := f.String("bucket", "", "bucket name")
+	access := f.String("access", "", "comma-separated actions: Read,Write,List,Tagging,Admin or none")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	if *bucket == "" {
+		return fmt.Errorf("-bucket is required")
+	}
+	if *access == "" {
+		return fmt.Errorf("-access is required")
+	}
+
+	return pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		// Get or create anonymous user
+		identity, isNew, err := getOrCreateAnonymousUser(ctx, client)
+		if err != nil {
+			return err
+		}
+
+		// Remove existing actions for this bucket
+		var kept []string
+		for _, a := range identity.Actions {
+			parts := strings.SplitN(a, ":", 2)
+			if len(parts) != 2 || parts[1] != *bucket {
+				kept = append(kept, a)
+			}
+		}
+
+		// Add new actions unless "none"
+		canonicalActions := map[string]string{
+			"read": "Read", "write": "Write", "list": "List",
+			"tagging": "Tagging", "admin": "Admin",
+		}
+		if strings.ToLower(strings.TrimSpace(*access)) != "none" {
+			seen := make(map[string]struct{})
+			for _, action := range strings.Split(*access, ",") {
+				action = strings.TrimSpace(action)
+				if action != "" {
+					canonical, ok := canonicalActions[strings.ToLower(action)]
+					if !ok {
+						return fmt.Errorf("invalid action %q: supported actions are Read, Write, List, Tagging, Admin", action)
+					}
+					if _, dup := seen[canonical]; dup {
+						continue
+					}
+					seen[canonical] = struct{}{}
+					kept = append(kept, canonical+":"+*bucket)
+				}
+			}
+		}
+
+		identity.Actions = kept
+
+		if isNew {
+			_, err = client.CreateUser(ctx, &iam_pb.CreateUserRequest{Identity: identity})
+		} else {
+			_, err = client.UpdateUser(ctx, &iam_pb.UpdateUserRequest{
+				Username: anonymousUserName,
+				Identity: identity,
+			})
+		}
+		if err != nil {
+			return err
+		}
+
+		fmt.Fprintf(writer, "Set anonymous access on bucket %q to: %s\n", *bucket, *access)
+		return nil
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+}
+
+func getOrCreateAnonymousUser(ctx context.Context, client iam_pb.SeaweedIdentityAccessManagementClient) (*iam_pb.Identity, bool, error) {
+	resp, err := client.GetUser(ctx, &iam_pb.GetUserRequest{Username: anonymousUserName})
+	if err == nil {
+		if resp.Identity == nil {
+			return nil, false, fmt.Errorf("anonymous user returned nil identity")
+		}
+		return resp.Identity, false, nil
+	}
+
+	st, ok := status.FromError(err)
+	if ok && st != nil && st.Code() == codes.NotFound {
+		return &iam_pb.Identity{
+			Name:    anonymousUserName,
+			Actions: []string{},
+		}, true, nil
+	}
+
+	return nil, false, fmt.Errorf("failed to get anonymous user: %w", err)
+}

weed/shell/command_s3_bucket_access.go

@@ -61,8 +61,8 @@ func (c *commandS3BucketAccess) Help() string {
 `
 }
 
-func (c *commandS3BucketAccess) HasTag(CommandTag) bool {
-	return false
+func (c *commandS3BucketAccess) HasTag(tag CommandTag) bool {
+	return tag == Hidden
 }
 
 func (c *commandS3BucketAccess) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {

weed/shell/command_s3_config_show.go

@@ -0,0 +1,116 @@
+package shell
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"text/tabwriter"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3ConfigShow{})
+}
+
+type commandS3ConfigShow struct {
+}
+
+func (c *commandS3ConfigShow) Name() string {
+	return "s3.config.show"
+}
+
+func (c *commandS3ConfigShow) Help() string {
+	return `show a summary of the current S3 IAM configuration
+
+	s3.config.show
+
+	Displays counts and a brief listing of users, policies, service accounts,
+	and groups. Use s3.iam.export for the full JSON dump.
+`
+}
+
+func (c *commandS3ConfigShow) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3ConfigShow) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	return pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		resp, err := client.GetConfiguration(ctx, &iam_pb.GetConfigurationRequest{})
+		if err != nil {
+			return err
+		}
+		cfg := resp.Configuration
+		if cfg == nil {
+			fmt.Fprintln(writer, "No S3 IAM configuration found.")
+			return nil
+		}
+
+		fmt.Fprintf(writer, "S3 IAM Configuration Summary\n")
+		fmt.Fprintf(writer, "============================\n\n")
+
+		// Users
+		fmt.Fprintf(writer, "Users: %d\n", len(cfg.Identities))
+		if len(cfg.Identities) > 0 {
+			tw := tabwriter.NewWriter(writer, 0, 4, 2, ' ', 0)
+			fmt.Fprintln(tw, "  NAME\tSTATUS\tSOURCE\tKEYS\tPOLICIES")
+			for _, id := range cfg.Identities {
+				status := "enabled"
+				if id.Disabled {
+					status = "disabled"
+				}
+				source := "dynamic"
+				if id.IsStatic {
+					source = "static"
+				}
+				policies := "-"
+				if len(id.PolicyNames) > 0 {
+					policies = joinMax(id.PolicyNames, 3)
+				}
+				fmt.Fprintf(tw, "  %s\t%s\t%s\t%d\t%s\n",
+					id.Name, status, source, len(id.Credentials), policies)
+			}
+			tw.Flush()
+		}
+		fmt.Fprintln(writer)
+
+		// Policies
+		fmt.Fprintf(writer, "Policies: %d\n", len(cfg.Policies))
+		if len(cfg.Policies) > 0 {
+			for _, p := range cfg.Policies {
+				fmt.Fprintf(writer, "  %s\n", p.Name)
+			}
+		}
+		fmt.Fprintln(writer)
+
+		// Service Accounts
+		fmt.Fprintf(writer, "Service Accounts: %d\n", len(cfg.ServiceAccounts))
+		if len(cfg.ServiceAccounts) > 0 {
+			for _, sa := range cfg.ServiceAccounts {
+				status := "enabled"
+				if sa.Disabled {
+					status = "disabled"
+				}
+				fmt.Fprintf(writer, "  %s (parent: %s, %s)\n", sa.Id, sa.ParentUser, status)
+			}
+		}
+		fmt.Fprintln(writer)
+
+		// Groups
+		fmt.Fprintf(writer, "Groups: %d\n", len(cfg.Groups))
+		if len(cfg.Groups) > 0 {
+			for _, g := range cfg.Groups {
+				fmt.Fprintf(writer, "  %s (%d members)\n", g.Name, len(g.Members))
+			}
+		}
+
+		return nil
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+}

weed/shell/command_s3_configure.go

@@ -31,21 +31,16 @@ func (c *commandS3Configure) Name() string {
 func (c *commandS3Configure) Help() string {
 	return `configure and apply s3 options for each bucket
 
-		# for workflow-oriented user management, prefer:
-		s3.user.create -name=username -generate_credentials
-		s3.user.show -name=username
-		s3.policy.attach -user=username -name=mypolicy
-
-		# see the current configuration file content
-		s3.configure
+	# see the current configuration file content
+	s3.configure
 
 	# create a new identity with account information
 	s3.configure -user=username -actions=Read,Write,List,Tagging -buckets=bucket-name -policies=policy1,policy2 -access_key=key -secret_key=secret -account_id=id -account_display_name=name -account_email=email@example.com -apply
 	`
 }
 
-func (c *commandS3Configure) HasTag(CommandTag) bool {
-	return false
+func (c *commandS3Configure) HasTag(tag CommandTag) bool {
+	return tag == Hidden
 }
 
 func (c *commandS3Configure) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {

weed/shell/command_s3_iam_export.go

@@ -0,0 +1,80 @@
+package shell
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"os"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3IAMExport{})
+}
+
+type commandS3IAMExport struct {
+}
+
+func (c *commandS3IAMExport) Name() string {
+	return "s3.iam.export"
+}
+
+func (c *commandS3IAMExport) Help() string {
+	return `export the full S3 IAM configuration as JSON
+
+	s3.iam.export
+	s3.iam.export -file backup.json
+
+	Exports all users, credentials, policies, service accounts, and groups.
+	Without -file, prints to stdout.
+`
+}
+
+func (c *commandS3IAMExport) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3IAMExport) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	file := f.String("file", "", "output file path (stdout if omitted)")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	return pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		resp, err := client.GetConfiguration(ctx, &iam_pb.GetConfigurationRequest{})
+		if err != nil {
+			return err
+		}
+
+		var out io.Writer = writer
+		if *file != "" {
+			fp, err := os.OpenFile(*file, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600)
+			if err != nil {
+				return fmt.Errorf("create file: %v", err)
+			}
+			defer fp.Close()
+			out = fp
+		}
+
+		if err := filer.ProtoToText(out, resp.Configuration); err != nil {
+			return err
+		}
+		fmt.Fprintln(out)
+
+		if *file != "" {
+			fmt.Fprintf(writer, "Exported IAM configuration to %s\n", *file)
+		}
+		return nil
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+}

weed/shell/command_s3_iam_import.go

@@ -0,0 +1,88 @@
+package shell
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"os"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3IAMImport{})
+}
+
+type commandS3IAMImport struct {
+}
+
+func (c *commandS3IAMImport) Name() string {
+	return "s3.iam.import"
+}
+
+func (c *commandS3IAMImport) Help() string {
+	return `import S3 IAM configuration from a JSON file
+
+	s3.iam.import -file backup.json -apply
+
+	Replaces the entire IAM configuration (users, credentials, policies,
+	service accounts, groups) with the contents of the file.
+
+	Requires -apply to confirm, since this overwrites the current configuration.
+`
+}
+
+func (c *commandS3IAMImport) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3IAMImport) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	file := f.String("file", "", "input JSON file")
+	apply := f.Bool("apply", false, "confirm overwrite of the entire IAM configuration")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	if *file == "" {
+		return fmt.Errorf("-file is required")
+	}
+	if !*apply {
+		return fmt.Errorf("this overwrites the entire IAM configuration; use -apply to confirm")
+	}
+
+	data, err := os.ReadFile(*file)
+	if err != nil {
+		return fmt.Errorf("read file: %w", err)
+	}
+
+	config := &iam_pb.S3ApiConfiguration{}
+	if err := filer.ParseS3ConfigurationFromBytes(data, config); err != nil {
+		return fmt.Errorf("parse configuration: %w", err)
+	}
+
+	err = pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+		_, err := client.PutConfiguration(ctx, &iam_pb.PutConfigurationRequest{
+			Configuration: config,
+		})
+		return err
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+	if err != nil {
+		return fmt.Errorf("put IAM configuration: %w", err)
+	}
+
+	fmt.Fprintf(writer, "Imported IAM configuration from %s\n", *file)
+	fmt.Fprintf(writer, "  Users:            %d\n", len(config.Identities))
+	fmt.Fprintf(writer, "  Policies:         %d\n", len(config.Policies))
+	fmt.Fprintf(writer, "  Service Accounts: %d\n", len(config.ServiceAccounts))
+	fmt.Fprintf(writer, "  Groups:           %d\n", len(config.Groups))
+	return nil
+}

weed/shell/command_s3_policy.go

@@ -29,18 +29,10 @@ func (c *commandS3Policy) Name() string {
 func (c *commandS3Policy) Help() string {
 	return `manage s3 policies
 
-		# preferred exact commands
-		s3.policy.create -name=mypolicy -file=policy.json
-		s3.policy.show -name=mypolicy
-		s3.policy.list
-		s3.policy.delete -name=mypolicy
-		s3.policy.attach -user=alice -name=mypolicy
-		s3.policy.detach -user=alice -name=mypolicy
+	# create or update a policy
+	s3.policy -put -name=mypolicy -file=policy.json
 
-		# create or update a policy
-		s3.policy -put -name=mypolicy -file=policy.json
-
-		# list all policies
+	# list all policies
 	s3.policy -list
 
 	# get a policy

weed/shell/command_s3_policy_attach.go

@@ -0,0 +1,93 @@
+package shell
+
+import (
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3PolicyAttach{})
+}
+
+type commandS3PolicyAttach struct {
+}
+
+func (c *commandS3PolicyAttach) Name() string {
+	return "s3.policy.attach"
+}
+
+func (c *commandS3PolicyAttach) Help() string {
+	return `attach a policy to an S3 IAM user
+
+	s3.policy.attach -policy <policy_name> -user <username>
+
+	The policy must already exist (create it with s3.policy -put).
+`
+}
+
+func (c *commandS3PolicyAttach) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3PolicyAttach) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	policy := f.String("policy", "", "policy name")
+	user := f.String("user", "", "user name")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	if *policy == "" {
+		return fmt.Errorf("-policy is required")
+	}
+	if *user == "" {
+		return fmt.Errorf("-user is required")
+	}
+
+	return pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		// Verify the policy exists
+		_, err := client.GetPolicy(ctx, &iam_pb.GetPolicyRequest{Name: *policy})
+		if err != nil {
+			return fmt.Errorf("get policy %q: %w", *policy, err)
+		}
+
+		// Get the user
+		resp, err := client.GetUser(ctx, &iam_pb.GetUserRequest{Username: *user})
+		if err != nil {
+			return fmt.Errorf("get user %q: %w", *user, err)
+		}
+		if resp.Identity == nil {
+			return fmt.Errorf("user %q returned empty identity", *user)
+		}
+
+		// Check if already attached
+		for _, p := range resp.Identity.PolicyNames {
+			if p == *policy {
+				return json.NewEncoder(writer).Encode(map[string]string{"policy": *policy, "user": *user})
+			}
+		}
+
+		resp.Identity.PolicyNames = append(resp.Identity.PolicyNames, *policy)
+		_, err = client.UpdateUser(ctx, &iam_pb.UpdateUserRequest{
+			Username: *user,
+			Identity: resp.Identity,
+		})
+		if err != nil {
+			return err
+		}
+
+		return json.NewEncoder(writer).Encode(map[string]string{"policy": *policy, "user": *user})
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+}

weed/shell/command_s3_policy_detach.go

@@ -0,0 +1,90 @@
+package shell
+
+import (
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3PolicyDetach{})
+}
+
+type commandS3PolicyDetach struct {
+}
+
+func (c *commandS3PolicyDetach) Name() string {
+	return "s3.policy.detach"
+}
+
+func (c *commandS3PolicyDetach) Help() string {
+	return `detach a policy from an S3 IAM user
+
+	s3.policy.detach -policy <policy_name> -user <username>
+`
+}
+
+func (c *commandS3PolicyDetach) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3PolicyDetach) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	policy := f.String("policy", "", "policy name")
+	user := f.String("user", "", "user name")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	if *policy == "" {
+		return fmt.Errorf("-policy is required")
+	}
+	if *user == "" {
+		return fmt.Errorf("-user is required")
+	}
+
+	return pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		resp, err := client.GetUser(ctx, &iam_pb.GetUserRequest{Username: *user})
+		if err != nil {
+			return fmt.Errorf("get user %q: %w", *user, err)
+		}
+		if resp.Identity == nil {
+			return fmt.Errorf("user %q returned empty identity", *user)
+		}
+
+		found := false
+		var kept []string
+		for _, p := range resp.Identity.PolicyNames {
+			if p == *policy {
+				found = true
+			} else {
+				kept = append(kept, p)
+			}
+		}
+		if !found {
+			return fmt.Errorf("policy %q is not attached to user %q", *policy, *user)
+		}
+
+		resp.Identity.PolicyNames = kept
+		_, err = client.UpdateUser(ctx, &iam_pb.UpdateUserRequest{
+			Username: *user,
+			Identity: resp.Identity,
+		})
+		if err != nil {
+			return err
+		}
+
+		return json.NewEncoder(writer).Encode(map[string]string{"policy": *policy, "user": *user})
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+}

weed/shell/command_s3_policy_subcommands.go

@@ -1,282 +0,0 @@
-package shell
-
-import (
-	"context"
-	"encoding/json"
-	"errors"
-	"flag"
-	"fmt"
-	"io"
-	"os"
-	"sort"
-	"strings"
-
-	"github.com/seaweedfs/seaweedfs/weed/credential"
-	"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
-)
-
-func init() {
-	Commands = append(Commands,
-		&s3ShellCommand{
-			name: "s3.policy.create",
-			help: `create or update a managed S3 policy
-
-	s3.policy.create -name photos-rw -file policy.json`,
-			do: runS3PolicyCreateCommand,
-		},
-		&s3ShellCommand{
-			name: "s3.policy.show",
-			help: `show one managed S3 policy
-
-	s3.policy.show -name photos-rw`,
-			do: runS3PolicyShowCommand,
-		},
-		&s3ShellCommand{
-			name: "s3.policy.list",
-			help: `list managed S3 policies`,
-			do:   runS3PolicyListCommand,
-		},
-		&s3ShellCommand{
-			name: "s3.policy.delete",
-			help: `delete a managed S3 policy
-
-	s3.policy.delete -name photos-rw`,
-			do: runS3PolicyDeleteCommand,
-		},
-		&s3ShellCommand{
-			name: "s3.policy.attach",
-			help: `attach a managed S3 policy to a filer-backed user
-
-	s3.policy.attach -user alice -name photos-rw`,
-			do: runS3PolicyAttachCommand,
-		},
-		&s3ShellCommand{
-			name: "s3.policy.detach",
-			help: `detach a managed S3 policy from a filer-backed user
-
-	s3.policy.detach -user alice -name photos-rw`,
-			do: runS3PolicyDetachCommand,
-		},
-	)
-}
-
-func runS3PolicyCreateCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
-	fs := flag.NewFlagSet("s3.policy.create", flag.ContinueOnError)
-	fs.SetOutput(writer)
-	name := fs.String("name", "", "policy name")
-	file := fs.String("file", "", "policy file (json)")
-	if err := fs.Parse(args); err != nil {
-		if err == flag.ErrHelp {
-			return nil
-		}
-		return err
-	}
-	if *name == "" {
-		return fmt.Errorf("-name is required")
-	}
-	if *file == "" {
-		return fmt.Errorf("-file is required")
-	}
-
-	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
-		return runS3PolicyCreate(ctx, store, *name, *file, writer)
-	})
-}
-
-func runS3PolicyShowCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
-	fs := flag.NewFlagSet("s3.policy.show", flag.ContinueOnError)
-	fs.SetOutput(writer)
-	name := fs.String("name", "", "policy name")
-	if err := fs.Parse(args); err != nil {
-		if err == flag.ErrHelp {
-			return nil
-		}
-		return err
-	}
-	if *name == "" {
-		return fmt.Errorf("-name is required")
-	}
-
-	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
-		return runS3PolicyShow(ctx, store, *name, writer)
-	})
-}
-
-func runS3PolicyListCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
-	fs := flag.NewFlagSet("s3.policy.list", flag.ContinueOnError)
-	fs.SetOutput(writer)
-	if err := fs.Parse(args); err != nil {
-		if err == flag.ErrHelp {
-			return nil
-		}
-		return err
-	}
-
-	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
-		return runS3PolicyList(ctx, store, writer)
-	})
-}
-
-func runS3PolicyDeleteCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
-	fs := flag.NewFlagSet("s3.policy.delete", flag.ContinueOnError)
-	fs.SetOutput(writer)
-	name := fs.String("name", "", "policy name")
-	if err := fs.Parse(args); err != nil {
-		if err == flag.ErrHelp {
-			return nil
-		}
-		return err
-	}
-	if *name == "" {
-		return fmt.Errorf("-name is required")
-	}
-
-	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
-		return runS3PolicyDelete(ctx, store, *name, writer)
-	})
-}
-
-func runS3PolicyAttachCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
-	return runS3PolicyAttachDetachCommand("s3.policy.attach", true, args, commandEnv, writer)
-}
-
-func runS3PolicyDetachCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
-	return runS3PolicyAttachDetachCommand("s3.policy.detach", false, args, commandEnv, writer)
-}
-
-func runS3PolicyAttachDetachCommand(commandName string, attach bool, args []string, commandEnv *CommandEnv, writer io.Writer) error {
-	fs := flag.NewFlagSet(commandName, flag.ContinueOnError)
-	fs.SetOutput(writer)
-	user := fs.String("user", "", "user name")
-	name := fs.String("name", "", "policy name")
-	if err := fs.Parse(args); err != nil {
-		if err == flag.ErrHelp {
-			return nil
-		}
-		return err
-	}
-	if *user == "" {
-		return fmt.Errorf("-user is required")
-	}
-	if *name == "" {
-		return fmt.Errorf("-name is required")
-	}
-
-	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
-		if attach {
-			return runS3PolicyAttach(ctx, store, *user, *name, writer)
-		}
-		return runS3PolicyDetach(ctx, store, *user, *name, writer)
-	})
-}
-
-func runS3PolicyCreate(ctx context.Context, store s3ShellStore, name, file string, writer io.Writer) error {
-	document, err := loadPolicyDocumentFromFile(file)
-	if err != nil {
-		return err
-	}
-	if err := store.CreatePolicy(ctx, name, *document); err != nil {
-		return err
-	}
-	fmt.Fprintf(writer, "Saved policy %q.\n", name)
-	return nil
-}
-
-func runS3PolicyShow(ctx context.Context, store s3ShellStore, name string, writer io.Writer) error {
-	document, err := store.GetPolicy(ctx, name)
-	if err != nil {
-		return err
-	}
-	if document == nil {
-		return fmt.Errorf("policy %q not found", name)
-	}
-	formatted, err := marshalPolicyDocument(document)
-	if err != nil {
-		return err
-	}
-	fmt.Fprintln(writer, formatted)
-	return nil
-}
-
-func runS3PolicyList(ctx context.Context, store s3ShellStore, writer io.Writer) error {
-	policies, err := store.GetPolicies(ctx)
-	if err != nil {
-		return err
-	}
-	names := make([]string, 0, len(policies))
-	for name := range policies {
-		names = append(names, name)
-	}
-	sort.Strings(names)
-	for _, name := range names {
-		fmt.Fprintln(writer, name)
-	}
-	return nil
-}
-
-func runS3PolicyDelete(ctx context.Context, store s3ShellStore, name string, writer io.Writer) error {
-	if err := store.DeletePolicy(ctx, name); err != nil {
-		if s3ShellIsNotFound(err) {
-			return fmt.Errorf("policy %q not found", name)
-		}
-		return err
-	}
-	fmt.Fprintf(writer, "Deleted policy %q.\n", name)
-	return nil
-}
-
-func runS3PolicyAttach(ctx context.Context, store s3ShellStore, username, policyName string, writer io.Writer) error {
-	identity, err := getS3User(ctx, store, username)
-	if err != nil {
-		return err
-	}
-	if err := ensureMutableIdentity(identity, "attach policies to"); err != nil {
-		return err
-	}
-	if err := store.AttachUserPolicy(ctx, username, policyName); err != nil {
-		if errors.Is(err, credential.ErrPolicyNotFound) {
-			return fmt.Errorf("policy %q not found", policyName)
-		}
-		return err
-	}
-	fmt.Fprintf(writer, "Attached policy %q to user %q.\n", policyName, username)
-	return nil
-}
-
-func runS3PolicyDetach(ctx context.Context, store s3ShellStore, username, policyName string, writer io.Writer) error {
-	identity, err := getS3User(ctx, store, username)
-	if err != nil {
-		return err
-	}
-	if err := ensureMutableIdentity(identity, "detach policies from"); err != nil {
-		return err
-	}
-	if err := store.DetachUserPolicy(ctx, username, policyName); err != nil {
-		if errors.Is(err, credential.ErrPolicyNotAttached) {
-			return fmt.Errorf("policy %q is not attached to user %q", policyName, username)
-		}
-		return err
-	}
-	fmt.Fprintf(writer, "Detached policy %q from user %q.\n", policyName, username)
-	return nil
-}
-
-func loadPolicyDocumentFromFile(file string) (*policy_engine.PolicyDocument, error) {
-	data, err := os.ReadFile(file)
-	if err != nil {
-		return nil, fmt.Errorf("failed to read policy file: %v", err)
-	}
-	var document policy_engine.PolicyDocument
-	if err := json.Unmarshal(data, &document); err != nil {
-		return nil, fmt.Errorf("invalid policy json: %v", err)
-	}
-	return &document, nil
-}
-
-func marshalPolicyDocument(document *policy_engine.PolicyDocument) (string, error) {
-	data, err := json.MarshalIndent(document, "", "  ")
-	if err != nil {
-		return "", err
-	}
-	return strings.TrimSpace(string(data)), nil
-}

weed/shell/command_s3_serviceaccount_create.go

@@ -0,0 +1,132 @@
+package shell
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"strings"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/iam"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3ServiceAccountCreate{})
+}
+
+type commandS3ServiceAccountCreate struct {
+}
+
+func (c *commandS3ServiceAccountCreate) Name() string {
+	return "s3.serviceaccount.create"
+}
+
+func (c *commandS3ServiceAccountCreate) Help() string {
+	return `create a service account for an S3 IAM user
+
+	s3.serviceaccount.create -user <parent_user> -description "my app"
+	s3.serviceaccount.create -user <parent_user> -actions Read,List -expiry 24h
+
+	Service accounts are linked to a parent user and can have restricted
+	permissions (a subset of the parent's actions).
+`
+}
+
+func (c *commandS3ServiceAccountCreate) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3ServiceAccountCreate) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	user := f.String("user", "", "parent user name")
+	description := f.String("description", "", "optional description")
+	actions := f.String("actions", "", "comma-separated actions (subset of parent)")
+	expiry := f.Duration("expiry", 0, "expiration duration (e.g. 24h, 0 = no expiration)")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	if *user == "" {
+		return fmt.Errorf("-user is required")
+	}
+
+	ak, err := iam.GenerateRandomString(iam.AccessKeyIdLength, iam.CharsetUpper)
+	if err != nil {
+		return fmt.Errorf("generate access key: %v", err)
+	}
+	sk, err := iam.GenerateSecretAccessKey()
+	if err != nil {
+		return fmt.Errorf("generate secret key: %v", err)
+	}
+
+	sa := &iam_pb.ServiceAccount{
+		ParentUser:  *user,
+		Description: *description,
+		Credential: &iam_pb.Credential{
+			AccessKey: ak,
+			SecretKey: sk,
+			Status:    iam.AccessKeyStatusActive,
+		},
+		CreatedAt: time.Now().Unix(),
+	}
+
+	validActions := map[string]string{
+		"read": "Read", "write": "Write", "list": "List",
+		"tagging": "Tagging", "admin": "Admin",
+	}
+	if *actions != "" {
+		seen := make(map[string]struct{})
+		for _, a := range strings.Split(*actions, ",") {
+			a = strings.TrimSpace(a)
+			if a != "" {
+				canonical, ok := validActions[strings.ToLower(a)]
+				if !ok {
+					return fmt.Errorf("invalid action %q: supported actions are Read, Write, List, Tagging, Admin", a)
+				}
+				if _, dup := seen[canonical]; dup {
+					continue
+				}
+				seen[canonical] = struct{}{}
+				sa.Actions = append(sa.Actions, canonical)
+			}
+		}
+	}
+
+	if *expiry < 0 {
+		return fmt.Errorf("-expiry must be >= 0")
+	}
+	if *expiry > 0 {
+		sa.Expiration = time.Now().Add(*expiry).Unix()
+	}
+
+	err = pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+		_, err := client.CreateServiceAccount(ctx, &iam_pb.CreateServiceAccountRequest{
+			ServiceAccount: sa,
+		})
+		return err
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+	if err != nil {
+		return err
+	}
+
+	fmt.Fprintf(writer, "Created service account for user %q\n", *user)
+	fmt.Fprintln(writer, "Note: use s3.serviceaccount.list to find the server-assigned ID.")
+	fmt.Fprintf(writer, "Access Key: %s\n", ak)
+	fmt.Fprintf(writer, "Secret Key: %s\n", sk)
+	if *description != "" {
+		fmt.Fprintf(writer, "Desc:       %s\n", *description)
+	}
+	if *expiry > 0 {
+		fmt.Fprintf(writer, "Expires:    %s\n", time.Unix(sa.Expiration, 0).Format(time.RFC3339))
+	}
+	fmt.Fprintln(writer)
+	fmt.Fprintln(writer, "Save these credentials - the secret key cannot be retrieved later.")
+	return nil
+}

weed/shell/command_s3_serviceaccount_delete.go

@@ -0,0 +1,61 @@
+package shell
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3ServiceAccountDelete{})
+}
+
+type commandS3ServiceAccountDelete struct {
+}
+
+func (c *commandS3ServiceAccountDelete) Name() string {
+	return "s3.serviceaccount.delete"
+}
+
+func (c *commandS3ServiceAccountDelete) Help() string {
+	return `delete a service account
+
+	s3.serviceaccount.delete -id <service_account_id>
+`
+}
+
+func (c *commandS3ServiceAccountDelete) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3ServiceAccountDelete) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	id := f.String("id", "", "service account ID")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	if *id == "" {
+		return fmt.Errorf("-id is required")
+	}
+
+	err := pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+		_, err := client.DeleteServiceAccount(ctx, &iam_pb.DeleteServiceAccountRequest{Id: *id})
+		return err
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+	if err != nil {
+		return err
+	}
+
+	fmt.Fprintf(writer, "Deleted service account %q\n", *id)
+	return nil
+}

weed/shell/command_s3_serviceaccount_list.go

@@ -0,0 +1,85 @@
+package shell
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"text/tabwriter"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3ServiceAccountList{})
+}
+
+type commandS3ServiceAccountList struct {
+}
+
+func (c *commandS3ServiceAccountList) Name() string {
+	return "s3.serviceaccount.list"
+}
+
+func (c *commandS3ServiceAccountList) Help() string {
+	return `list service accounts
+
+	s3.serviceaccount.list
+	s3.serviceaccount.list -user <parent_user>
+
+	Lists all service accounts, optionally filtered by parent user.
+`
+}
+
+func (c *commandS3ServiceAccountList) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3ServiceAccountList) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	user := f.String("user", "", "filter by parent user (optional)")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	return pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		resp, err := client.ListServiceAccounts(ctx, &iam_pb.ListServiceAccountsRequest{})
+		if err != nil {
+			return err
+		}
+
+		var filtered []*iam_pb.ServiceAccount
+		for _, sa := range resp.ServiceAccounts {
+			if *user == "" || sa.ParentUser == *user {
+				filtered = append(filtered, sa)
+			}
+		}
+
+		if len(filtered) == 0 {
+			fmt.Fprintln(writer, "No service accounts found.")
+			return nil
+		}
+
+		tw := tabwriter.NewWriter(writer, 0, 4, 2, ' ', 0)
+		fmt.Fprintln(tw, "ID\tPARENT\tSTATUS\tDESCRIPTION")
+		for _, sa := range filtered {
+			st := "enabled"
+			if sa.Disabled {
+				st = "disabled"
+			}
+			desc := sa.Description
+			if len(desc) > 40 {
+				desc = desc[:37] + "..."
+			}
+			fmt.Fprintf(tw, "%s\t%s\t%s\t%s\n", sa.Id, sa.ParentUser, st, desc)
+		}
+		return tw.Flush()
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+}

weed/shell/command_s3_serviceaccount_show.go

@@ -0,0 +1,93 @@
+package shell
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"strings"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3ServiceAccountShow{})
+}
+
+type commandS3ServiceAccountShow struct {
+}
+
+func (c *commandS3ServiceAccountShow) Name() string {
+	return "s3.serviceaccount.show"
+}
+
+func (c *commandS3ServiceAccountShow) Help() string {
+	return `show details of a service account
+
+	s3.serviceaccount.show -id <service_account_id>
+`
+}
+
+func (c *commandS3ServiceAccountShow) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3ServiceAccountShow) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	id := f.String("id", "", "service account ID")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	if *id == "" {
+		return fmt.Errorf("-id is required")
+	}
+
+	return pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		resp, err := client.GetServiceAccount(ctx, &iam_pb.GetServiceAccountRequest{Id: *id})
+		if err != nil {
+			return err
+		}
+		sa := resp.ServiceAccount
+
+		status := "enabled"
+		if sa.Disabled {
+			status = "disabled"
+		}
+
+		fmt.Fprintf(writer, "ID:          %s\n", sa.Id)
+		fmt.Fprintf(writer, "Parent:      %s\n", sa.ParentUser)
+		fmt.Fprintf(writer, "Status:      %s\n", status)
+		if sa.Description != "" {
+			fmt.Fprintf(writer, "Description: %s\n", sa.Description)
+		}
+		if sa.Credential != nil {
+			st := sa.Credential.Status
+			if st == "" {
+				st = "Active"
+			}
+			fmt.Fprintf(writer, "Access Key:  %s (%s)\n", sa.Credential.AccessKey, st)
+		}
+		if len(sa.Actions) > 0 {
+			fmt.Fprintf(writer, "Actions:     %s\n", strings.Join(sa.Actions, ", "))
+		}
+		if sa.Expiration > 0 {
+			fmt.Fprintf(writer, "Expires:     %s\n", time.Unix(sa.Expiration, 0).Format(time.RFC3339))
+		}
+		if sa.CreatedAt > 0 {
+			fmt.Fprintf(writer, "Created:     %s\n", time.Unix(sa.CreatedAt, 0).Format(time.RFC3339))
+		}
+		if sa.CreatedBy != "" {
+			fmt.Fprintf(writer, "Created By:  %s\n", sa.CreatedBy)
+		}
+
+		return nil
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+}

weed/shell/command_s3_user_create.go

@@ -0,0 +1,111 @@
+package shell
+
+import (
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"os"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/iam"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3UserCreate{})
+}
+
+type commandS3UserCreate struct {
+}
+
+func (c *commandS3UserCreate) Name() string {
+	return "s3.user.create"
+}
+
+func (c *commandS3UserCreate) Help() string {
+	return `create an S3 IAM user
+
+	s3.user.create -name <username>
+	s3.user.create -name <username> -access_key <key> -secret_key <secret>
+
+	Creates a new user with a credential pair. If -access_key and -secret_key
+	are omitted, they are generated automatically.
+
+	After creating a user, attach policies with s3.policy.attach.
+
+	Output: JSON to stdout. Secret key is printed to stderr only.
+`
+}
+
+func (c *commandS3UserCreate) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3UserCreate) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	name := f.String("name", "", "user name")
+	accessKey := f.String("access_key", "", "access key (generated if omitted)")
+	secretKey := f.String("secret_key", "", "secret key (generated if omitted)")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	if *name == "" {
+		return fmt.Errorf("-name is required")
+	}
+
+	ak := *accessKey
+	sk := *secretKey
+	generated := false
+
+	if ak == "" && sk == "" {
+		generated = true
+		var err error
+		ak, err = iam.GenerateRandomString(iam.AccessKeyIdLength, iam.CharsetUpper)
+		if err != nil {
+			return fmt.Errorf("generate access key: %v", err)
+		}
+		sk, err = iam.GenerateSecretAccessKey()
+		if err != nil {
+			return fmt.Errorf("generate secret key: %v", err)
+		}
+	} else if ak == "" || sk == "" {
+		return fmt.Errorf("both -access_key and -secret_key must be provided together, or omit both to auto-generate")
+	}
+
+	identity := &iam_pb.Identity{
+		Name: *name,
+		Credentials: []*iam_pb.Credential{
+			{
+				AccessKey: ak,
+				SecretKey: sk,
+				Status:    iam.AccessKeyStatusActive,
+			},
+		},
+	}
+
+	err := pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+		_, err := client.CreateUser(ctx, &iam_pb.CreateUserRequest{Identity: identity})
+		return err
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+	if err != nil {
+		return err
+	}
+
+	if generated {
+		fmt.Fprintf(os.Stderr, "Secret Key: %s\n", sk)
+		fmt.Fprintf(os.Stderr, "Save this secret key - it cannot be retrieved later.\n")
+	}
+
+	return json.NewEncoder(writer).Encode(map[string]string{
+		"name":       *name,
+		"access_key": ak,
+	})
+}

weed/shell/command_s3_user_delete.go

@@ -0,0 +1,61 @@
+package shell
+
+import (
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3UserDelete{})
+}
+
+type commandS3UserDelete struct {
+}
+
+func (c *commandS3UserDelete) Name() string {
+	return "s3.user.delete"
+}
+
+func (c *commandS3UserDelete) Help() string {
+	return `delete an S3 IAM user
+
+	s3.user.delete -name <username>
+`
+}
+
+func (c *commandS3UserDelete) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3UserDelete) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	name := f.String("name", "", "user name")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	if *name == "" {
+		return fmt.Errorf("-name is required")
+	}
+
+	err := pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+		_, err := client.DeleteUser(ctx, &iam_pb.DeleteUserRequest{Username: *name})
+		return err
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+	if err != nil {
+		return err
+	}
+
+	return json.NewEncoder(writer).Encode(map[string]string{"name": *name})
+}

weed/shell/command_s3_user_disable.go

@@ -0,0 +1,81 @@
+package shell
+
+import (
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3UserDisable{})
+}
+
+type commandS3UserDisable struct {
+}
+
+func (c *commandS3UserDisable) Name() string {
+	return "s3.user.disable"
+}
+
+func (c *commandS3UserDisable) Help() string {
+	return `disable an S3 IAM user
+
+	s3.user.disable -name <username>
+
+	Disabled users cannot authenticate. Their credentials and policies
+	are preserved and will take effect again when the user is re-enabled.
+`
+}
+
+func (c *commandS3UserDisable) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3UserDisable) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	name := f.String("name", "", "user name")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	if *name == "" {
+		return fmt.Errorf("-name is required")
+	}
+
+	err := pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		resp, err := client.GetUser(ctx, &iam_pb.GetUserRequest{Username: *name})
+		if err != nil {
+			return fmt.Errorf("get user %q: %w", *name, err)
+		}
+		if resp.Identity == nil {
+			return fmt.Errorf("user %q returned empty identity", *name)
+		}
+
+		if resp.Identity.Disabled {
+			return nil
+		}
+
+		resp.Identity.Disabled = true
+		_, err = client.UpdateUser(ctx, &iam_pb.UpdateUserRequest{
+			Username: *name,
+			Identity: resp.Identity,
+		})
+		return err
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+	if err != nil {
+		return err
+	}
+
+	return json.NewEncoder(writer).Encode(map[string]string{"name": *name, "status": "disabled"})
+}

weed/shell/command_s3_user_enable.go

@@ -0,0 +1,77 @@
+package shell
+
+import (
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3UserEnable{})
+}
+
+type commandS3UserEnable struct {
+}
+
+func (c *commandS3UserEnable) Name() string {
+	return "s3.user.enable"
+}
+
+func (c *commandS3UserEnable) Help() string {
+	return `enable a disabled S3 IAM user
+
+	s3.user.enable -name <username>
+`
+}
+
+func (c *commandS3UserEnable) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3UserEnable) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	name := f.String("name", "", "user name")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	if *name == "" {
+		return fmt.Errorf("-name is required")
+	}
+
+	err := pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		resp, err := client.GetUser(ctx, &iam_pb.GetUserRequest{Username: *name})
+		if err != nil {
+			return fmt.Errorf("get user %q: %w", *name, err)
+		}
+		if resp.Identity == nil {
+			return fmt.Errorf("user %q returned empty identity", *name)
+		}
+
+		if resp.Identity.Disabled {
+			resp.Identity.Disabled = false
+			_, err = client.UpdateUser(ctx, &iam_pb.UpdateUserRequest{
+				Username: *name,
+				Identity: resp.Identity,
+			})
+			return err
+		}
+		return nil
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+	if err != nil {
+		return err
+	}
+
+	return json.NewEncoder(writer).Encode(map[string]string{"name": *name, "status": "enabled"})
+}

weed/shell/command_s3_user_list.go

@@ -0,0 +1,87 @@
+package shell
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"strings"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3UserList{})
+}
+
+type commandS3UserList struct {
+}
+
+func (c *commandS3UserList) Name() string {
+	return "s3.user.list"
+}
+
+func (c *commandS3UserList) Help() string {
+	return `list S3 IAM users
+
+	s3.user.list
+
+	Output: JSON array of users with status, policies, and credential count.
+`
+}
+
+func (c *commandS3UserList) HasTag(CommandTag) bool {
+	return false
+}
+
+type s3UserListEntry struct {
+	Name     string   `json:"name"`
+	Status   string   `json:"status"`
+	Policies []string `json:"policies"`
+	Keys     int      `json:"keys"`
+}
+
+func (c *commandS3UserList) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	return pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		resp, err := client.GetConfiguration(ctx, &iam_pb.GetConfigurationRequest{})
+		if err != nil {
+			return err
+		}
+
+		var result []s3UserListEntry
+		for _, id := range resp.Configuration.GetIdentities() {
+			status := "enabled"
+			if id.Disabled {
+				status = "disabled"
+			}
+			policies := id.PolicyNames
+			if policies == nil {
+				policies = []string{}
+			}
+			result = append(result, s3UserListEntry{
+				Name:     id.Name,
+				Status:   status,
+				Policies: policies,
+				Keys:     len(id.Credentials),
+			})
+		}
+		if result == nil {
+			result = []s3UserListEntry{}
+		}
+		return json.NewEncoder(writer).Encode(result)
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+}
+
+// joinMax joins up to max strings with ", " and appends "..." if truncated.
+func joinMax(items []string, max int) string {
+	if len(items) <= max {
+		return strings.Join(items, ", ")
+	}
+	return strings.Join(items[:max], ", ") + "..."
+}

weed/shell/command_s3_user_provision.go

@@ -0,0 +1,179 @@
+package shell
+
+import (
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"strings"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/iam"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3UserProvision{})
+}
+
+type commandS3UserProvision struct {
+}
+
+func (c *commandS3UserProvision) Name() string {
+	return "s3.user.provision"
+}
+
+func (c *commandS3UserProvision) Help() string {
+	return `create a user with a bucket policy in one step
+
+	s3.user.provision -name <username> -bucket <bucket_name> -role readwrite
+	s3.user.provision -name <username> -bucket <bucket_name> -role readonly
+
+	Convenience wrapper that performs these steps:
+	  1. Creates an IAM policy for the bucket and role
+	  2. Creates the user with auto-generated credentials
+	  3. Attaches the policy to the user
+
+	Roles:
+	  readonly   - s3:GetObject, s3:ListBucket
+	  readwrite  - s3:GetObject, s3:PutObject, s3:DeleteObject, s3:ListBucket
+	  admin      - s3:* (full access to the bucket)
+`
+}
+
+func (c *commandS3UserProvision) HasTag(CommandTag) bool {
+	return false
+}
+
+var rolePolicies = map[string][]string{
+	"readonly":  {"s3:GetObject"},
+	"readwrite": {"s3:GetObject", "s3:PutObject", "s3:DeleteObject"},
+	"admin":     {"s3:*"},
+}
+
+func (c *commandS3UserProvision) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	name := f.String("name", "", "user name")
+	bucket := f.String("bucket", "", "bucket name")
+	role := f.String("role", "", "role: readonly, readwrite, or admin")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	if *name == "" {
+		return fmt.Errorf("-name is required")
+	}
+	if *bucket == "" {
+		return fmt.Errorf("-bucket is required")
+	}
+	if strings.ContainsAny(*bucket, "*?") {
+		return fmt.Errorf("-bucket must be a literal bucket name, not a wildcard pattern")
+	}
+	if *role == "" {
+		return fmt.Errorf("-role is required (readonly, readwrite, admin)")
+	}
+
+	actions, ok := rolePolicies[*role]
+	if !ok {
+		return fmt.Errorf("unknown role %q: must be readonly, readwrite, or admin", *role)
+	}
+
+	policyName := fmt.Sprintf("%s-%s-%s", *bucket, *name, *role)
+
+	// Build the policy document
+	bucketActions := []string{"s3:ListBucket"}
+	if *role == "admin" {
+		bucketActions = []string{"s3:*"}
+	}
+	policyDoc := map[string]interface{}{
+		"Version": "2012-10-17",
+		"Statement": []map[string]interface{}{
+			{
+				"Effect":   "Allow",
+				"Action":   actions,
+				"Resource": []string{fmt.Sprintf("arn:aws:s3:::%s/*", *bucket)},
+			},
+			{
+				"Effect":   "Allow",
+				"Action":   bucketActions,
+				"Resource": []string{fmt.Sprintf("arn:aws:s3:::%s", *bucket)},
+			},
+		},
+	}
+	policyJSON, err := json.Marshal(policyDoc)
+	if err != nil {
+		return fmt.Errorf("marshal policy: %v", err)
+	}
+
+	// Generate credentials
+	ak, err := iam.GenerateRandomString(iam.AccessKeyIdLength, iam.CharsetUpper)
+	if err != nil {
+		return fmt.Errorf("generate access key: %v", err)
+	}
+	sk, err := iam.GenerateSecretAccessKey()
+	if err != nil {
+		return fmt.Errorf("generate secret key: %v", err)
+	}
+
+	err = pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		// Step 0: Check if user already exists
+		if resp, getErr := client.GetUser(ctx, &iam_pb.GetUserRequest{Username: *name}); getErr == nil && resp.Identity != nil {
+			return fmt.Errorf("user %q already exists", *name)
+		} else if getErr != nil && status.Code(getErr) != codes.NotFound {
+			return fmt.Errorf("check user existence: %w", getErr)
+		}
+
+		// Step 1: Create policy
+		_, err := client.PutPolicy(ctx, &iam_pb.PutPolicyRequest{
+			Name:    policyName,
+			Content: string(policyJSON),
+		})
+		if err != nil {
+			return fmt.Errorf("create policy: %v", err)
+		}
+		fmt.Fprintf(writer, "Created policy %q\n", policyName)
+
+		// Step 2: Create user
+		identity := &iam_pb.Identity{
+			Name: *name,
+			Credentials: []*iam_pb.Credential{
+				{
+					AccessKey: ak,
+					SecretKey: sk,
+					Status:    iam.AccessKeyStatusActive,
+				},
+			},
+			PolicyNames: []string{policyName},
+		}
+		_, err = client.CreateUser(ctx, &iam_pb.CreateUserRequest{Identity: identity})
+		if err != nil {
+			// Rollback: remove the policy we just created
+			if _, delErr := client.DeletePolicy(ctx, &iam_pb.DeletePolicyRequest{Name: policyName}); delErr != nil {
+				fmt.Fprintf(writer, "Warning: failed to rollback policy %q: %v\n", policyName, delErr)
+			}
+			return fmt.Errorf("create user: %w", err)
+		}
+		fmt.Fprintf(writer, "Created user %q with policy %q attached\n", *name, policyName)
+
+		return nil
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+	if err != nil {
+		return err
+	}
+
+	fmt.Fprintln(writer)
+	fmt.Fprintf(writer, "Access Key: %s\n", ak)
+	fmt.Fprintf(writer, "Secret Key: %s\n", sk)
+	fmt.Fprintln(writer)
+	fmt.Fprintln(writer, "Save these credentials - the secret key cannot be retrieved later.")
+	return nil
+}

weed/shell/command_s3_user_show.go

@@ -0,0 +1,137 @@
+package shell
+
+import (
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3UserShow{})
+}
+
+type commandS3UserShow struct {
+}
+
+func (c *commandS3UserShow) Name() string {
+	return "s3.user.show"
+}
+
+func (c *commandS3UserShow) Help() string {
+	return `show details of an S3 IAM user
+
+	s3.user.show -name <username>
+
+	Output: JSON object with user details.
+`
+}
+
+func (c *commandS3UserShow) HasTag(CommandTag) bool {
+	return false
+}
+
+type s3CredentialInfo struct {
+	AccessKey string `json:"access_key"`
+	Status    string `json:"status"`
+}
+
+type s3AccountInfo struct {
+	ID          string `json:"id,omitempty"`
+	DisplayName string `json:"display_name,omitempty"`
+	Email       string `json:"email,omitempty"`
+}
+
+type s3UserShowResult struct {
+	Name            string            `json:"name"`
+	Status          string            `json:"status"`
+	Source          string            `json:"source"`
+	Account         *s3AccountInfo    `json:"account,omitempty"`
+	Policies        []string          `json:"policies"`
+	Actions         []string          `json:"actions,omitempty"`
+	Credentials     []s3CredentialInfo `json:"credentials"`
+	ServiceAccounts []string          `json:"service_accounts,omitempty"`
+}
+
+func (c *commandS3UserShow) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
+	f := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	name := f.String("name", "", "user name")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+
+	if *name == "" {
+		return fmt.Errorf("-name is required")
+	}
+
+	return pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		resp, err := client.GetUser(ctx, &iam_pb.GetUserRequest{Username: *name})
+		if err != nil {
+			return err
+		}
+		id := resp.Identity
+		if id == nil {
+			return fmt.Errorf("user %q returned empty identity", *name)
+		}
+
+		status := "enabled"
+		if id.Disabled {
+			status = "disabled"
+		}
+		source := "dynamic"
+		if id.IsStatic {
+			source = "static"
+		}
+
+		result := s3UserShowResult{
+			Name:   id.Name,
+			Status: status,
+			Source: source,
+		}
+
+		if id.Account != nil && (id.Account.Id != "" || id.Account.DisplayName != "" || id.Account.EmailAddress != "") {
+			result.Account = &s3AccountInfo{
+				ID:          id.Account.Id,
+				DisplayName: id.Account.DisplayName,
+				Email:       id.Account.EmailAddress,
+			}
+		}
+
+		result.Policies = id.PolicyNames
+		if result.Policies == nil {
+			result.Policies = []string{}
+		}
+
+		if len(id.Actions) > 0 {
+			result.Actions = id.Actions
+		}
+
+		result.Credentials = make([]s3CredentialInfo, 0, len(id.Credentials))
+		for _, cred := range id.Credentials {
+			st := cred.Status
+			if st == "" {
+				st = "Active"
+			}
+			result.Credentials = append(result.Credentials, s3CredentialInfo{
+				AccessKey: cred.AccessKey,
+				Status:    st,
+			})
+		}
+
+		if len(id.ServiceAccountIds) > 0 {
+			result.ServiceAccounts = id.ServiceAccountIds
+		}
+
+		return json.NewEncoder(writer).Encode(result)
+	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)
+}

weed/shell/command_s3_user_subcommands.go

@@ -1,712 +0,0 @@
-package shell
-
-import (
-	"context"
-	"crypto/rand"
-	"errors"
-	"flag"
-	"fmt"
-	"io"
-	"sort"
-	"strings"
-	"text/tabwriter"
-	"time"
-
-	"github.com/seaweedfs/seaweedfs/weed/credential"
-	iamgrpc "github.com/seaweedfs/seaweedfs/weed/credential/grpc"
-	weediam "github.com/seaweedfs/seaweedfs/weed/iam"
-	"github.com/seaweedfs/seaweedfs/weed/pb"
-	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
-	"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
-	"google.golang.org/protobuf/proto"
-)
-
-var errS3AccessKeyInUse = errors.New("access key already in use")
-
-type s3ShellCommand struct {
-	name string
-	help string
-	do   func([]string, *CommandEnv, io.Writer) error
-}
-
-func (c *s3ShellCommand) Name() string { return c.name }
-
-func (c *s3ShellCommand) Help() string { return c.help }
-
-func (c *s3ShellCommand) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
-	return c.do(args, commandEnv, writer)
-}
-
-func (c *s3ShellCommand) HasTag(CommandTag) bool { return false }
-
-type s3ShellStore interface {
-	CreateUser(ctx context.Context, identity *iam_pb.Identity) error
-	GetUser(ctx context.Context, username string) (*iam_pb.Identity, error)
-	UpdateUser(ctx context.Context, username string, identity *iam_pb.Identity) error
-	DeleteUser(ctx context.Context, username string) error
-	ListUsers(ctx context.Context) ([]string, error)
-	GetUserByAccessKey(ctx context.Context, accessKey string) (*iam_pb.Identity, error)
-	CreateAccessKey(ctx context.Context, username string, credential *iam_pb.Credential) error
-	DeleteAccessKey(ctx context.Context, username string, accessKey string) error
-	GetPolicy(ctx context.Context, name string) (*policy_engine.PolicyDocument, error)
-	GetPolicies(ctx context.Context) (map[string]policy_engine.PolicyDocument, error)
-	CreatePolicy(ctx context.Context, name string, document policy_engine.PolicyDocument) error
-	DeletePolicy(ctx context.Context, name string) error
-	AttachUserPolicy(ctx context.Context, username string, policyName string) error
-	DetachUserPolicy(ctx context.Context, username string, policyName string) error
-	ListAttachedUserPolicies(ctx context.Context, username string) ([]string, error)
-}
-
-type s3UserCreateOptions struct {
-	name                string
-	accessKey           string
-	secretKey           string
-	generateCredentials bool
-	accountID           string
-	displayName         string
-	email               string
-}
-
-type s3AccessKeyCreateOptions struct {
-	username            string
-	accessKey           string
-	secretKey           string
-	generateCredentials bool
-}
-
-func init() {
-	Commands = append(Commands,
-		&s3ShellCommand{
-			name: "s3.user.list",
-			help: `list S3 users with source and status`,
-			do:   runS3UserListCommand,
-		},
-		&s3ShellCommand{
-			name: "s3.user.show",
-			help: `show one S3 user's details
-
-	s3.user.show -name alice`,
-			do: runS3UserShowCommand,
-		},
-		&s3ShellCommand{
-			name: "s3.user.create",
-			help: `create a filer-backed S3 user
-
-	s3.user.create -name alice -generate_credentials
-	s3.user.create -name alice -access_key AKIA... -secret_key secret
-	s3.user.create -name alice -email alice@example.com -display_name "Alice"`,
-			do: runS3UserCreateCommand,
-		},
-		&s3ShellCommand{
-			name: "s3.user.delete",
-			help: `delete a filer-backed S3 user
-
-	s3.user.delete -name alice`,
-			do: runS3UserDeleteCommand,
-		},
-		&s3ShellCommand{
-			name: "s3.user.enable",
-			help: `enable a filer-backed S3 user
-
-	s3.user.enable -name alice`,
-			do: runS3UserEnableCommand,
-		},
-		&s3ShellCommand{
-			name: "s3.user.disable",
-			help: `disable a filer-backed S3 user
-
-	s3.user.disable -name alice`,
-			do: runS3UserDisableCommand,
-		},
-		&s3ShellCommand{
-			name: "s3.user.accesskey.list",
-			help: `list access keys for a filer-backed S3 user
-
-	s3.user.accesskey.list -user alice`,
-			do: runS3UserAccessKeyListCommand,
-		},
-		&s3ShellCommand{
-			name: "s3.user.accesskey.create",
-			help: `create an access key for a filer-backed S3 user
-
-	s3.user.accesskey.create -user alice -generate_credentials
-	s3.user.accesskey.create -user alice -access_key AKIA... -secret_key secret`,
-			do: runS3UserAccessKeyCreateCommand,
-		},
-		&s3ShellCommand{
-			name: "s3.user.accesskey.delete",
-			help: `delete an access key for a filer-backed S3 user
-
-	s3.user.accesskey.delete -user alice -access_key AKIA...`,
-			do: runS3UserAccessKeyDeleteCommand,
-		},
-	)
-}
-
-func runS3UserListCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
-	fs := flag.NewFlagSet("s3.user.list", flag.ContinueOnError)
-	fs.SetOutput(writer)
-	if err := fs.Parse(args); err != nil {
-		if err == flag.ErrHelp {
-			return nil
-		}
-		return err
-	}
-
-	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
-		return runS3UserList(ctx, store, writer)
-	})
-}
-
-func runS3UserShowCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
-	fs := flag.NewFlagSet("s3.user.show", flag.ContinueOnError)
-	fs.SetOutput(writer)
-	name := fs.String("name", "", "user name")
-	if err := fs.Parse(args); err != nil {
-		if err == flag.ErrHelp {
-			return nil
-		}
-		return err
-	}
-	if *name == "" {
-		return fmt.Errorf("-name is required")
-	}
-
-	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
-		return runS3UserShow(ctx, store, *name, writer)
-	})
-}
-
-func runS3UserCreateCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
-	fs := flag.NewFlagSet("s3.user.create", flag.ContinueOnError)
-	fs.SetOutput(writer)
-	opts := s3UserCreateOptions{}
-	fs.StringVar(&opts.name, "name", "", "user name")
-	fs.StringVar(&opts.accessKey, "access_key", "", "access key to create")
-	fs.StringVar(&opts.secretKey, "secret_key", "", "secret key to create")
-	fs.BoolVar(&opts.generateCredentials, "generate_credentials", false, "generate an initial access key pair")
-	fs.StringVar(&opts.accountID, "account_id", "", "account id")
-	fs.StringVar(&opts.displayName, "display_name", "", "account display name")
-	fs.StringVar(&opts.email, "email", "", "account email")
-	if err := fs.Parse(args); err != nil {
-		if err == flag.ErrHelp {
-			return nil
-		}
-		return err
-	}
-	if opts.name == "" {
-		return fmt.Errorf("-name is required")
-	}
-
-	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
-		return runS3UserCreate(ctx, store, opts, writer)
-	})
-}
-
-func runS3UserDeleteCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
-	fs := flag.NewFlagSet("s3.user.delete", flag.ContinueOnError)
-	fs.SetOutput(writer)
-	name := fs.String("name", "", "user name")
-	if err := fs.Parse(args); err != nil {
-		if err == flag.ErrHelp {
-			return nil
-		}
-		return err
-	}
-	if *name == "" {
-		return fmt.Errorf("-name is required")
-	}
-
-	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
-		return runS3UserDelete(ctx, store, *name, writer)
-	})
-}
-
-func runS3UserEnableCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
-	return runS3UserStatusCommand("s3.user.enable", false, args, commandEnv, writer)
-}
-
-func runS3UserDisableCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
-	return runS3UserStatusCommand("s3.user.disable", true, args, commandEnv, writer)
-}
-
-func runS3UserStatusCommand(commandName string, disabled bool, args []string, commandEnv *CommandEnv, writer io.Writer) error {
-	fs := flag.NewFlagSet(commandName, flag.ContinueOnError)
-	fs.SetOutput(writer)
-	name := fs.String("name", "", "user name")
-	if err := fs.Parse(args); err != nil {
-		if err == flag.ErrHelp {
-			return nil
-		}
-		return err
-	}
-	if *name == "" {
-		return fmt.Errorf("-name is required")
-	}
-
-	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
-		return runS3UserSetDisabled(ctx, store, *name, disabled, writer)
-	})
-}
-
-func runS3UserAccessKeyListCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
-	fs := flag.NewFlagSet("s3.user.accesskey.list", flag.ContinueOnError)
-	fs.SetOutput(writer)
-	user := fs.String("user", "", "user name")
-	if err := fs.Parse(args); err != nil {
-		if err == flag.ErrHelp {
-			return nil
-		}
-		return err
-	}
-	if *user == "" {
-		return fmt.Errorf("-user is required")
-	}
-
-	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
-		return runS3UserAccessKeyList(ctx, store, *user, writer)
-	})
-}
-
-func runS3UserAccessKeyCreateCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
-	fs := flag.NewFlagSet("s3.user.accesskey.create", flag.ContinueOnError)
-	fs.SetOutput(writer)
-	opts := s3AccessKeyCreateOptions{}
-	fs.StringVar(&opts.username, "user", "", "user name")
-	fs.StringVar(&opts.accessKey, "access_key", "", "access key to create")
-	fs.StringVar(&opts.secretKey, "secret_key", "", "secret key to create")
-	fs.BoolVar(&opts.generateCredentials, "generate_credentials", false, "generate an access key pair")
-	if err := fs.Parse(args); err != nil {
-		if err == flag.ErrHelp {
-			return nil
-		}
-		return err
-	}
-	if opts.username == "" {
-		return fmt.Errorf("-user is required")
-	}
-
-	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
-		return runS3UserAccessKeyCreate(ctx, store, opts, writer)
-	})
-}
-
-func runS3UserAccessKeyDeleteCommand(args []string, commandEnv *CommandEnv, writer io.Writer) error {
-	fs := flag.NewFlagSet("s3.user.accesskey.delete", flag.ContinueOnError)
-	fs.SetOutput(writer)
-	user := fs.String("user", "", "user name")
-	accessKey := fs.String("access_key", "", "access key to delete")
-	if err := fs.Parse(args); err != nil {
-		if err == flag.ErrHelp {
-			return nil
-		}
-		return err
-	}
-	if *user == "" {
-		return fmt.Errorf("-user is required")
-	}
-	if *accessKey == "" {
-		return fmt.Errorf("-access_key is required")
-	}
-
-	return withS3ShellStore(commandEnv, func(ctx context.Context, store s3ShellStore) error {
-		return runS3UserAccessKeyDelete(ctx, store, *user, *accessKey, writer)
-	})
-}
-
-func runS3UserList(ctx context.Context, store s3ShellStore, writer io.Writer) error {
-	usernames, err := store.ListUsers(ctx)
-	if err != nil {
-		return err
-	}
-	sort.Strings(usernames)
-
-	tw := tabwriter.NewWriter(writer, 0, 0, 2, ' ', 0)
-	fmt.Fprintln(tw, "NAME\tSOURCE\tSTATUS\tACCESS KEYS\tPOLICIES")
-	for _, username := range usernames {
-		identity, err := store.GetUser(ctx, username)
-		if err != nil {
-			return err
-		}
-		fmt.Fprintf(tw, "%s\t%s\t%s\t%d\t%d\n",
-			identity.Name,
-			s3IdentitySource(identity),
-			s3IdentityStatus(identity),
-			len(identity.Credentials),
-			len(identity.PolicyNames),
-		)
-	}
-	return tw.Flush()
-}
-
-func runS3UserShow(ctx context.Context, store s3ShellStore, username string, writer io.Writer) error {
-	identity, err := getS3User(ctx, store, username)
-	if err != nil {
-		return err
-	}
-	writeS3IdentityDetails(writer, identity)
-	return nil
-}
-
-func runS3UserCreate(ctx context.Context, store s3ShellStore, opts s3UserCreateOptions, writer io.Writer) error {
-	existing, err := store.GetUser(ctx, opts.name)
-	switch {
-	case err == nil && existing != nil:
-		if existing.IsStatic {
-			return fmt.Errorf("user %q already exists in -s3.config; edit the static config for bootstrap users", opts.name)
-		}
-		return fmt.Errorf("user %q already exists", opts.name)
-	case err != nil && !s3ShellIsNotFound(err):
-		return err
-	}
-
-	identity := &iam_pb.Identity{Name: opts.name}
-	if opts.accountID != "" || opts.displayName != "" || opts.email != "" {
-		identity.Account = &iam_pb.Account{
-			Id:           opts.accountID,
-			DisplayName:  opts.displayName,
-			EmailAddress: opts.email,
-		}
-	}
-
-	createdCredential, err := s3BuildCredentialForCreate(ctx, store, opts.accessKey, opts.secretKey, opts.generateCredentials || opts.accessKey != "" || opts.secretKey != "")
-	if err != nil {
-		return err
-	}
-	if createdCredential != nil {
-		identity.Credentials = []*iam_pb.Credential{createdCredential}
-	}
-
-	if err := store.CreateUser(ctx, identity); err != nil {
-		if s3ShellIsAlreadyExists(err) {
-			return fmt.Errorf("user %q already exists", opts.name)
-		}
-		return err
-	}
-
-	fmt.Fprintf(writer, "Created user %q.\n", opts.name)
-	if createdCredential != nil {
-		writeCreatedCredential(writer, createdCredential)
-	}
-	return nil
-}
-
-func runS3UserDelete(ctx context.Context, store s3ShellStore, username string, writer io.Writer) error {
-	identity, err := getS3User(ctx, store, username)
-	if err != nil {
-		return err
-	}
-	if err := ensureMutableIdentity(identity, "delete"); err != nil {
-		return err
-	}
-	if err := store.DeleteUser(ctx, username); err != nil {
-		if s3ShellIsNotFound(err) {
-			return fmt.Errorf("user %q not found", username)
-		}
-		return err
-	}
-	fmt.Fprintf(writer, "Deleted user %q.\n", username)
-	return nil
-}
-
-func runS3UserSetDisabled(ctx context.Context, store s3ShellStore, username string, disabled bool, writer io.Writer) error {
-	identity, err := getS3User(ctx, store, username)
-	if err != nil {
-		return err
-	}
-	if err := ensureMutableIdentity(identity, ternary(disabled, "disable", "enable")); err != nil {
-		return err
-	}
-	if identity.Disabled == disabled {
-		fmt.Fprintf(writer, "User %q is already %s.\n", username, s3IdentityStatus(identity))
-		return nil
-	}
-
-	updated := proto.Clone(identity).(*iam_pb.Identity)
-	updated.Disabled = disabled
-	if err := store.UpdateUser(ctx, username, updated); err != nil {
-		return err
-	}
-	if disabled {
-		fmt.Fprintf(writer, "Disabled user %q.\n", username)
-	} else {
-		fmt.Fprintf(writer, "Enabled user %q.\n", username)
-	}
-	return nil
-}
-
-func runS3UserAccessKeyList(ctx context.Context, store s3ShellStore, username string, writer io.Writer) error {
-	identity, err := getS3User(ctx, store, username)
-	if err != nil {
-		return err
-	}
-
-	tw := tabwriter.NewWriter(writer, 0, 0, 2, ' ', 0)
-	fmt.Fprintln(tw, "USER\tACCESS KEY\tSTATUS")
-	for _, credential := range sortedCredentials(identity.Credentials) {
-		fmt.Fprintf(tw, "%s\t%s\t%s\n", username, credential.AccessKey, s3CredentialStatus(credential))
-	}
-	if len(identity.Credentials) == 0 {
-		fmt.Fprintf(tw, "%s\t%s\t%s\n", username, "-", "-")
-	}
-	return tw.Flush()
-}
-
-func runS3UserAccessKeyCreate(ctx context.Context, store s3ShellStore, opts s3AccessKeyCreateOptions, writer io.Writer) error {
-	identity, err := getS3User(ctx, store, opts.username)
-	if err != nil {
-		return err
-	}
-	if err := ensureMutableIdentity(identity, "create access keys for"); err != nil {
-		return err
-	}
-
-	createdCredential, err := s3BuildCredentialForCreate(ctx, store, opts.accessKey, opts.secretKey, opts.generateCredentials || opts.accessKey != "" || opts.secretKey != "")
-	if err != nil {
-		return err
-	}
-	if createdCredential == nil {
-		return fmt.Errorf("set -generate_credentials or provide -access_key and/or -secret_key")
-	}
-
-	if err := store.CreateAccessKey(ctx, opts.username, createdCredential); err != nil {
-		if s3ShellIsAlreadyExists(err) {
-			return fmt.Errorf("access key %q is already in use", createdCredential.AccessKey)
-		}
-		return err
-	}
-
-	fmt.Fprintf(writer, "Created access key for user %q.\n", opts.username)
-	writeCreatedCredential(writer, createdCredential)
-	return nil
-}
-
-func runS3UserAccessKeyDelete(ctx context.Context, store s3ShellStore, username string, accessKey string, writer io.Writer) error {
-	identity, err := getS3User(ctx, store, username)
-	if err != nil {
-		return err
-	}
-	if err := ensureMutableIdentity(identity, "delete access keys for"); err != nil {
-		return err
-	}
-	if err := store.DeleteAccessKey(ctx, username, accessKey); err != nil {
-		if s3ShellIsNotFound(err) {
-			return fmt.Errorf("access key %q not found for user %q", accessKey, username)
-		}
-		return err
-	}
-	fmt.Fprintf(writer, "Deleted access key %q for user %q.\n", accessKey, username)
-	return nil
-}
-
-func withS3ShellStore(commandEnv *CommandEnv, fn func(ctx context.Context, store s3ShellStore) error) error {
-	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
-	defer cancel()
-
-	store := &iamgrpc.IamGrpcStore{}
-	store.SetFilerAddressFunc(func() pb.ServerAddress {
-		return commandEnv.option.FilerAddress
-	}, commandEnv.option.GrpcDialOption)
-
-	return fn(ctx, store)
-}
-
-func ensureMutableIdentity(identity *iam_pb.Identity, action string) error {
-	if identity != nil && identity.IsStatic {
-		return fmt.Errorf("cannot %s user %q because it comes from -s3.config; edit the static config for bootstrap users", action, identity.Name)
-	}
-	return nil
-}
-
-func getS3User(ctx context.Context, store s3ShellStore, username string) (*iam_pb.Identity, error) {
-	identity, err := store.GetUser(ctx, username)
-	if err != nil {
-		if s3ShellIsNotFound(err) {
-			return nil, fmt.Errorf("user %q not found", username)
-		}
-		return nil, err
-	}
-	return identity, nil
-}
-
-func s3BuildCredentialForCreate(ctx context.Context, store s3ShellStore, accessKey, secretKey string, shouldCreate bool) (*iam_pb.Credential, error) {
-	if !shouldCreate {
-		return nil, nil
-	}
-
-	if accessKey == "" {
-		var err error
-		accessKey, err = generateUniqueAccessKey(ctx, store)
-		if err != nil {
-			return nil, err
-		}
-	} else {
-		if err := ensureAccessKeyAvailable(ctx, store, accessKey); err != nil {
-			return nil, err
-		}
-	}
-	if secretKey == "" {
-		var err error
-		secretKey, err = weediam.GenerateSecretAccessKey()
-		if err != nil {
-			return nil, fmt.Errorf("generate secret key: %w", err)
-		}
-	}
-
-	return &iam_pb.Credential{
-		AccessKey: accessKey,
-		SecretKey: secretKey,
-		Status:    weediam.AccessKeyStatusActive,
-	}, nil
-}
-
-func ensureAccessKeyAvailable(ctx context.Context, store s3ShellStore, accessKey string) error {
-	_, err := store.GetUserByAccessKey(ctx, accessKey)
-	if err == nil {
-		return fmt.Errorf("%w: %s", errS3AccessKeyInUse, accessKey)
-	}
-	if s3ShellIsNotFound(err) {
-		return nil
-	}
-	return err
-}
-
-func generateUniqueAccessKey(ctx context.Context, store s3ShellStore) (string, error) {
-	for range 16 {
-		accessKey, err := generateAccessKey()
-		if err != nil {
-			return "", err
-		}
-		if err := ensureAccessKeyAvailable(ctx, store, accessKey); err == nil {
-			return accessKey, nil
-		} else if !errors.Is(err, errS3AccessKeyInUse) {
-			return "", err
-		}
-	}
-	return "", fmt.Errorf("failed to generate a unique access key")
-}
-
-func generateAccessKey() (string, error) {
-	const charset = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
-	var raw [20]byte
-	if _, err := rand.Read(raw[:]); err != nil {
-		return "", fmt.Errorf("generate access key: %w", err)
-	}
-	for i := range raw {
-		raw[i] = charset[int(raw[i])%len(charset)]
-	}
-	return string(raw[:]), nil
-}
-
-func s3IdentitySource(identity *iam_pb.Identity) string {
-	if identity != nil && identity.IsStatic {
-		return "static"
-	}
-	return "dynamic"
-}
-
-func s3IdentityStatus(identity *iam_pb.Identity) string {
-	if identity != nil && identity.Disabled {
-		return "disabled"
-	}
-	return "enabled"
-}
-
-func s3CredentialStatus(credential *iam_pb.Credential) string {
-	if credential == nil || credential.Status == "" {
-		return weediam.AccessKeyStatusActive
-	}
-	return credential.Status
-}
-
-func sortedCredentials(credentials []*iam_pb.Credential) []*iam_pb.Credential {
-	out := append([]*iam_pb.Credential(nil), credentials...)
-	sort.Slice(out, func(i, j int) bool {
-		return out[i].AccessKey < out[j].AccessKey
-	})
-	return out
-}
-
-func writeCreatedCredential(writer io.Writer, credential *iam_pb.Credential) {
-	fmt.Fprintf(writer, "Access Key: %s\n", credential.AccessKey)
-	fmt.Fprintf(writer, "Secret Key: %s\n", credential.SecretKey)
-}
-
-func writeS3IdentityDetails(writer io.Writer, identity *iam_pb.Identity) {
-	fmt.Fprintf(writer, "Name: %s\n", identity.Name)
-	fmt.Fprintf(writer, "Source: %s\n", s3IdentitySource(identity))
-	fmt.Fprintf(writer, "Status: %s\n", s3IdentityStatus(identity))
-	if identity.Account != nil {
-		if identity.Account.Id != "" {
-			fmt.Fprintf(writer, "Account ID: %s\n", identity.Account.Id)
-		}
-		if identity.Account.DisplayName != "" {
-			fmt.Fprintf(writer, "Display Name: %s\n", identity.Account.DisplayName)
-		}
-		if identity.Account.EmailAddress != "" {
-			fmt.Fprintf(writer, "Email: %s\n", identity.Account.EmailAddress)
-		}
-	}
-
-	if len(identity.Credentials) == 0 {
-		fmt.Fprintln(writer, "Access Keys: none")
-	} else {
-		fmt.Fprintln(writer, "Access Keys:")
-		for _, credential := range sortedCredentials(identity.Credentials) {
-			fmt.Fprintf(writer, "  %s (%s)\n", credential.AccessKey, s3CredentialStatus(credential))
-		}
-	}
-
-	if len(identity.PolicyNames) == 0 {
-		fmt.Fprintln(writer, "Policies: none")
-	} else {
-		policies := append([]string(nil), identity.PolicyNames...)
-		sort.Strings(policies)
-		fmt.Fprintf(writer, "Policies: %s\n", strings.Join(policies, ", "))
-	}
-
-	if len(identity.Actions) == 0 {
-		fmt.Fprintln(writer, "Actions: none")
-	} else {
-		actions := append([]string(nil), identity.Actions...)
-		sort.Strings(actions)
-		fmt.Fprintf(writer, "Actions: %s\n", strings.Join(actions, ", "))
-	}
-}
-
-func s3ShellIsNotFound(err error) bool {
-	if err == nil {
-		return false
-	}
-	if errors.Is(err, credential.ErrUserNotFound) ||
-		errors.Is(err, credential.ErrAccessKeyNotFound) ||
-		errors.Is(err, credential.ErrPolicyNotFound) {
-		return true
-	}
-	if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound {
-		return true
-	}
-	return false
-}
-
-func s3ShellIsAlreadyExists(err error) bool {
-	if err == nil {
-		return false
-	}
-	if errors.Is(err, credential.ErrUserAlreadyExists) {
-		return true
-	}
-	if st, ok := status.FromError(err); ok && st.Code() == codes.AlreadyExists {
-		return true
-	}
-	return false
-}
-
-func ternary[T any](cond bool, onTrue, onFalse T) T {
-	if cond {
-		return onTrue
-	}
-	return onFalse
-}

weed/shell/command_s3_user_subcommands_test.go

@@ -1,191 +0,0 @@
-package shell
-
-import (
-	"bytes"
-	"context"
-	"strings"
-	"testing"
-
-	"github.com/seaweedfs/seaweedfs/weed/credential/memory"
-	weediam "github.com/seaweedfs/seaweedfs/weed/iam"
-	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
-	"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
-)
-
-func newTestS3ShellStore(t *testing.T) *memory.MemoryStore {
-	t.Helper()
-
-	store := &memory.MemoryStore{}
-	if err := store.Initialize(nil, ""); err != nil {
-		t.Fatalf("initialize memory store: %v", err)
-	}
-	return store
-}
-
-func TestRunS3UserCreateGeneratesCredentials(t *testing.T) {
-	store := newTestS3ShellStore(t)
-
-	var out bytes.Buffer
-	err := runS3UserCreate(context.Background(), store, s3UserCreateOptions{
-		name:                "alice",
-		generateCredentials: true,
-		email:               "alice@example.com",
-	}, &out)
-	if err != nil {
-		t.Fatalf("runS3UserCreate: %v", err)
-	}
-
-	user, err := store.GetUser(context.Background(), "alice")
-	if err != nil {
-		t.Fatalf("GetUser: %v", err)
-	}
-	if len(user.Credentials) != 1 {
-		t.Fatalf("expected 1 credential, got %d", len(user.Credentials))
-	}
-	if got := user.Credentials[0].Status; got != weediam.AccessKeyStatusActive {
-		t.Fatalf("expected %q status, got %q", weediam.AccessKeyStatusActive, got)
-	}
-	if user.Account == nil || user.Account.EmailAddress != "alice@example.com" {
-		t.Fatalf("expected account email to be set, got %+v", user.Account)
-	}
-
-	output := out.String()
-	if !strings.Contains(output, `Created user "alice".`) {
-		t.Fatalf("expected created message, got %q", output)
-	}
-	if !strings.Contains(output, "Access Key:") || !strings.Contains(output, "Secret Key:") {
-		t.Fatalf("expected credential output, got %q", output)
-	}
-}
-
-func TestRunS3UserSetDisabledRejectsStaticUser(t *testing.T) {
-	store := newTestS3ShellStore(t)
-	if err := store.CreateUser(context.Background(), &iam_pb.Identity{
-		Name:     "bootstrap-admin",
-		IsStatic: true,
-	}); err != nil {
-		t.Fatalf("CreateUser: %v", err)
-	}
-
-	err := runS3UserSetDisabled(context.Background(), store, "bootstrap-admin", true, &bytes.Buffer{})
-	if err == nil {
-		t.Fatal("expected static user mutation to fail")
-	}
-	if !strings.Contains(err.Error(), "-s3.config") {
-		t.Fatalf("expected static user guidance, got %v", err)
-	}
-}
-
-func TestRunS3UserListIncludesSourceAndStatus(t *testing.T) {
-	store := newTestS3ShellStore(t)
-	if err := store.CreateUser(context.Background(), &iam_pb.Identity{
-		Name: "alice",
-		Credentials: []*iam_pb.Credential{
-			{AccessKey: "ALICEKEY", SecretKey: "secret", Status: weediam.AccessKeyStatusActive},
-		},
-	}); err != nil {
-		t.Fatalf("CreateUser alice: %v", err)
-	}
-	if err := store.CreateUser(context.Background(), &iam_pb.Identity{
-		Name:     "bootstrap-admin",
-		IsStatic: true,
-		Disabled: true,
-	}); err != nil {
-		t.Fatalf("CreateUser bootstrap-admin: %v", err)
-	}
-
-	var out bytes.Buffer
-	if err := runS3UserList(context.Background(), store, &out); err != nil {
-		t.Fatalf("runS3UserList: %v", err)
-	}
-
-	output := out.String()
-	if !strings.Contains(output, "NAME") || !strings.Contains(output, "SOURCE") || !strings.Contains(output, "STATUS") {
-		t.Fatalf("expected header, got %q", output)
-	}
-	if !strings.Contains(output, "alice") || !strings.Contains(output, "dynamic") || !strings.Contains(output, "enabled") {
-		t.Fatalf("expected dynamic user row, got %q", output)
-	}
-	if !strings.Contains(output, "bootstrap-admin") || !strings.Contains(output, "static") || !strings.Contains(output, "disabled") {
-		t.Fatalf("expected static user row, got %q", output)
-	}
-}
-
-func TestRunS3UserAccessKeyCreateGeneratesCredential(t *testing.T) {
-	store := newTestS3ShellStore(t)
-	if err := store.CreateUser(context.Background(), &iam_pb.Identity{Name: "alice"}); err != nil {
-		t.Fatalf("CreateUser: %v", err)
-	}
-
-	var out bytes.Buffer
-	err := runS3UserAccessKeyCreate(context.Background(), store, s3AccessKeyCreateOptions{
-		username:            "alice",
-		generateCredentials: true,
-	}, &out)
-	if err != nil {
-		t.Fatalf("runS3UserAccessKeyCreate: %v", err)
-	}
-
-	user, err := store.GetUser(context.Background(), "alice")
-	if err != nil {
-		t.Fatalf("GetUser: %v", err)
-	}
-	if len(user.Credentials) != 1 {
-		t.Fatalf("expected 1 credential, got %d", len(user.Credentials))
-	}
-	if !strings.Contains(out.String(), `Created access key for user "alice".`) {
-		t.Fatalf("expected create message, got %q", out.String())
-	}
-}
-
-func TestRunS3PolicyAttachUpdatesUserPolicies(t *testing.T) {
-	store := newTestS3ShellStore(t)
-	if err := store.CreateUser(context.Background(), &iam_pb.Identity{Name: "alice"}); err != nil {
-		t.Fatalf("CreateUser: %v", err)
-	}
-	if err := store.CreatePolicy(context.Background(), "photos-rw", policy_engine.PolicyDocument{
-		Version: "2012-10-17",
-	}); err != nil {
-		t.Fatalf("CreatePolicy: %v", err)
-	}
-
-	var out bytes.Buffer
-	if err := runS3PolicyAttach(context.Background(), store, "alice", "photos-rw", &out); err != nil {
-		t.Fatalf("runS3PolicyAttach: %v", err)
-	}
-
-	user, err := store.GetUser(context.Background(), "alice")
-	if err != nil {
-		t.Fatalf("GetUser: %v", err)
-	}
-	if len(user.PolicyNames) != 1 || user.PolicyNames[0] != "photos-rw" {
-		t.Fatalf("expected attached policy, got %v", user.PolicyNames)
-	}
-	if !strings.Contains(out.String(), `Attached policy "photos-rw" to user "alice".`) {
-		t.Fatalf("expected attach message, got %q", out.String())
-	}
-}
-
-func TestRunS3PolicyDetachRejectsStaticUser(t *testing.T) {
-	store := newTestS3ShellStore(t)
-	if err := store.CreateUser(context.Background(), &iam_pb.Identity{
-		Name:        "bootstrap-admin",
-		IsStatic:    true,
-		PolicyNames: []string{"photos-rw"},
-	}); err != nil {
-		t.Fatalf("CreateUser: %v", err)
-	}
-	if err := store.CreatePolicy(context.Background(), "photos-rw", policy_engine.PolicyDocument{
-		Version: "2012-10-17",
-	}); err != nil {
-		t.Fatalf("CreatePolicy: %v", err)
-	}
-
-	err := runS3PolicyDetach(context.Background(), store, "bootstrap-admin", "photos-rw", &bytes.Buffer{})
-	if err == nil {
-		t.Fatal("expected static user mutation to fail")
-	}
-	if !strings.Contains(err.Error(), "-s3.config") {
-		t.Fatalf("expected static user guidance, got %v", err)
-	}
-}

weed/shell/shell_liner.go

@@ -1,6 +1,7 @@
 package shell
 
 import (
+	"bufio"
 	"context"
 	"fmt"
 	"io"
@@ -64,31 +65,46 @@ func RunShell(options ShellOptions) {
 			}
 			return nil
 		})
-		fmt.Printf("master: %s ", *options.Masters)
+		fmt.Fprintf(os.Stderr, "master: %s ", *options.Masters)
 		if len(filers) > 0 {
-			fmt.Printf("filers: %v", filers)
+			fmt.Fprintf(os.Stderr, "filers: %v", filers)
 			commandEnv.option.FilerAddress = filers[rand.IntN(len(filers))]
 		}
-		fmt.Println()
+		fmt.Fprintln(os.Stderr)
 	}
 
-	for {
-		cmd, err := line.Prompt("> ")
-		if err != nil {
-			if err != io.EOF {
-				fmt.Printf("%v\n", err)
-			}
-			return
-		}
-
-		if strings.TrimSpace(cmd) != "" {
-			line.AppendHistory(cmd)
-		}
-
-		for _, c := range util.StringSplit(cmd, ";") {
-			if processEachCmd(c, commandEnv) {
+	if liner.TerminalSupported() {
+		for {
+			cmd, err := line.Prompt("> ")
+			if err != nil {
+				if err != io.EOF {
+					fmt.Fprintf(os.Stderr, "%v\n", err)
+				}
 				return
 			}
+
+			if strings.TrimSpace(cmd) != "" {
+				line.AppendHistory(cmd)
+			}
+
+			for _, c := range util.StringSplit(cmd, ";") {
+				if processEachCmd(c, commandEnv) {
+					return
+				}
+			}
+		}
+	} else {
+		scanner := bufio.NewScanner(os.Stdin)
+		for scanner.Scan() {
+			cmd := scanner.Text()
+			for _, c := range util.StringSplit(cmd, ";") {
+				if processEachCmd(c, commandEnv) {
+					return
+				}
+			}
+		}
+		if err := scanner.Err(); err != nil {
+			fmt.Fprintf(os.Stderr, "error reading stdin: %v\n", err)
 		}
 	}
 }
@@ -186,6 +202,9 @@ func printGenericHelp() {
 	fmt.Print(msg)
 
 	for _, c := range Commands {
+		if c.HasTag(Hidden) {
+			continue
+		}
 		helpTexts := strings.SplitN(c.Help(), "\n", 2)
 		fmt.Printf("  %-30s\t# %s \n", c.Name(), helpTexts[0])
 	}
@@ -229,10 +248,10 @@ func loadHistory() {
 
 func saveHistory() {
 	if f, err := os.Create(historyPath); err != nil {
-		fmt.Printf("Error creating history file: %v\n", err)
+		fmt.Fprintf(os.Stderr, "Error creating history file: %v\n", err)
 	} else {
 		if _, err = line.WriteHistory(f); err != nil {
-			fmt.Printf("Error writing history file: %v\n", err)
+			fmt.Fprintf(os.Stderr, "Error writing history file: %v\n", err)
 		}
 		f.Close()
 	}

weed/storage/erasure_coding/ec_consistency_test.go

@@ -0,0 +1,224 @@
+package erasure_coding
+
+import (
+	"bytes"
+	"crypto/rand"
+	"fmt"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+)
+
+// TestEcConsistency_WritesBetweenEncodeAndEcx reproduces a race condition that
+// existed in VolumeEcShardsGenerate before the fix in this PR.
+//
+// Previously, the order was:
+//   1. WriteEcFilesWithContext(baseFileName, ecCtx)        — EC shards from .dat
+//   2. WriteSortedFileFromIdx(v.IndexFileName(), ".ecx")   — .ecx from .idx
+//
+// If a write appended data to .dat/.idx between steps 1 and 2, the .ecx would
+// have entries pointing to data that doesn't exist in the EC shards.
+//
+// The fix reverses the order (write .ecx first, then generate EC shards), so
+// that .ecx is always a subset of what the EC shards contain.
+//
+// This test simulates the old buggy sequence to validate that the problem is real.
+func TestEcConsistency_WritesBetweenEncodeAndEcx(t *testing.T) {
+	dir := t.TempDir()
+	baseFileName := dir + "/consistency"
+
+	ctx := NewDefaultECContext("", 0)
+
+	// Phase 1: Create initial .dat and .idx with known data
+	datSize := int64(largeBlockSize*DataShardsCount + smallBlockSize*DataShardsCount*3) // 1 large row + 3 small rows
+	originalData := make([]byte, datSize)
+	rand.Read(originalData)
+
+	err := os.WriteFile(baseFileName+".dat", originalData, 0644)
+	require.NoError(t, err)
+
+	// Create a minimal .idx with one entry pointing to the data
+	createTestIdx(t, baseFileName+".idx", []idxEntry{
+		{id: 1, offset: 0, size: types.Size(datSize)},
+	})
+
+	// Phase 2: EC encode — generates .ec00-.ec13 from current .dat
+	err = generateEcFiles(baseFileName, int(smallBlockSize), largeBlockSize, smallBlockSize, ctx)
+	require.NoError(t, err, "EC encoding")
+
+	// Phase 3: SIMULATE a write between EC encoding and .ecx generation
+	// (reproducing the old buggy order where .ecx was generated after EC shards)
+	extraData := make([]byte, 5000)
+	rand.Read(extraData)
+
+	f, err := os.OpenFile(baseFileName+".dat", os.O_WRONLY|os.O_APPEND, 0644)
+	require.NoError(t, err)
+	_, err = f.Write(extraData)
+	require.NoError(t, err)
+	f.Close()
+
+	// Update .idx with the new entry
+	createTestIdx(t, baseFileName+".idx", []idxEntry{
+		{id: 1, offset: 0, size: types.Size(datSize)},
+		{id: 2, offset: datSize, size: types.Size(len(extraData))},
+	})
+
+	// Phase 4: Generate .ecx from the UPDATED .idx (as the old buggy code did)
+	err = WriteSortedFileFromIdx(baseFileName, ".ecx")
+	require.NoError(t, err, "WriteSortedFileFromIdx")
+
+	// Phase 5: Now try to read needle 2 via EC shards — it should fail
+	// because the EC shards were generated from the OLD .dat (without the extra data)
+	ecFiles, err := openEcFiles(baseFileName, true, ctx)
+	require.NoError(t, err)
+	defer closeEcFiles(ecFiles)
+
+	ecStat, err := ecFiles[0].Stat()
+	require.NoError(t, err)
+	shardSize := ecStat.Size()
+
+	// Read needle 2 (the one added after EC encoding) using LocateData.
+	// Use shardSize-1 to simulate the ecdFileSize fallback path used by
+	// LocateEcShardNeedleInterval when datFileSize is unavailable.
+	actualSize := needle.GetActualSize(types.Size(len(extraData)), needle.Version3)
+	intervals := LocateData(largeBlockSize, smallBlockSize, shardSize-1, datSize, types.Size(actualSize))
+
+	t.Logf("Trying to read needle 2 at offset %d size %d from EC shards (shardSize=%d)", datSize, actualSize, shardSize)
+	t.Logf("Intervals: %+v", intervals)
+
+	// Try to read — this will either fail with an error (offset out of bounds)
+	// or return garbage data (the padded zeros from EC encoding)
+	ecData, readErr := assembleFromIntervalsAllowError(ecFiles, intervals, largeBlockSize, smallBlockSize)
+
+	if readErr != nil {
+		t.Logf("CONFIRMED: Read error for needle written after EC encoding: %v", readErr)
+	} else {
+		// If we got data, it should be zeros (padding) or garbage, not the actual extraData
+		isAllZeros := true
+		for _, b := range ecData {
+			if b != 0 {
+				isAllZeros = false
+				break
+			}
+		}
+		if isAllZeros {
+			t.Logf("CONFIRMED: Read returned zero-padded data (EC shards don't have the needle)")
+		} else if !bytes.Equal(ecData[:len(extraData)], extraData) {
+			t.Logf("CONFIRMED: Read returned wrong data (EC shards don't have the needle)")
+		} else {
+			t.Error("UNEXPECTED: Read returned correct data — needle should NOT be in EC shards")
+		}
+	}
+
+	// Phase 6: Verify a small read from the original data still works.
+	// Use the correct shardDatSize (from the original datSize, not the modified one)
+	// to avoid the fallback heuristic issues.
+	shardDatSize := datSize / int64(DataShardsCount)
+	readSize := types.Size(smallBlockSize)
+	intervals1 := LocateData(largeBlockSize, smallBlockSize, shardDatSize, 0, readSize)
+	ecData1, err := assembleFromIntervalsAllowError(ecFiles, intervals1, largeBlockSize, smallBlockSize)
+	require.NoError(t, err, "reading original data from EC shards")
+
+	assert.True(t, bytes.Equal(originalData[:readSize], ecData1),
+		"Original data at offset 0 should match EC shard data")
+	t.Logf("Original data reads correctly from EC shards")
+}
+
+// TestEcConsistency_ExactLargeRowEncoding verifies that generateEcFiles correctly
+// encodes a .dat file whose size is exactly one large row (DataShardsCount *
+// largeBlockSize), producing shards of exactly largeBlockSize each, and that
+// every chunk of the encoded data can be read back correctly via LocateData.
+func TestEcConsistency_ExactLargeRowEncoding(t *testing.T) {
+	dir := t.TempDir()
+	baseFileName := dir + "/exact"
+	ctx := NewDefaultECContext("", 0)
+
+	datSize := int64(largeBlockSize * DataShardsCount) // exactly 1 large row
+	data := make([]byte, datSize)
+	rand.Read(data)
+	err := os.WriteFile(baseFileName+".dat", data, 0644)
+	require.NoError(t, err)
+
+	// EC encode
+	err = generateEcFiles(baseFileName, int(smallBlockSize), largeBlockSize, smallBlockSize, ctx)
+	require.NoError(t, err)
+
+	// Check shard sizes — each shard should be exactly largeBlockSize
+	ecFiles, err := openEcFiles(baseFileName, true, ctx)
+	require.NoError(t, err)
+	defer closeEcFiles(ecFiles)
+
+	for i := 0; i < ctx.DataShards; i++ {
+		stat, err := ecFiles[i].Stat()
+		require.NoError(t, err, "stat shard %d", i)
+		assert.Equal(t, int64(largeBlockSize), stat.Size(),
+			"data shard %d should be exactly largeBlockSize", i)
+	}
+
+	// Verify data reads correctly at every smallBlockSize offset via LocateData
+	shardDatSize := datSize / int64(ctx.DataShards)
+	readSize := types.Size(smallBlockSize)
+	for offset := int64(0); offset+int64(readSize) <= datSize; offset += int64(readSize) {
+		intervals := LocateData(largeBlockSize, smallBlockSize, shardDatSize, offset, readSize)
+		ecData, err := assembleFromIntervalsAllowError(ecFiles, intervals, largeBlockSize, smallBlockSize)
+		require.NoError(t, err, "reading at offset %d", offset)
+		expected := data[offset : offset+int64(readSize)]
+		assert.True(t, bytes.Equal(expected, ecData),
+			"data mismatch at offset %d", offset)
+	}
+}
+
+type idxEntry struct {
+	id     types.NeedleId
+	offset int64
+	size   types.Size
+}
+
+func createTestIdx(t *testing.T, filename string, entries []idxEntry) {
+	t.Helper()
+	f, err := os.Create(filename)
+	require.NoError(t, err)
+	defer f.Close()
+
+	buf := make([]byte, types.NeedleMapEntrySize)
+	for _, e := range entries {
+		types.NeedleIdToBytes(buf[:types.NeedleIdSize], e.id)
+		types.OffsetToBytes(buf[types.NeedleIdSize:types.NeedleIdSize+types.OffsetSize], types.ToOffset(e.offset))
+		types.SizeToBytes(buf[types.NeedleIdSize+types.OffsetSize:], e.size)
+		_, err := f.Write(buf)
+		require.NoError(t, err)
+	}
+}
+
+func assembleFromIntervalsAllowError(ecFiles []*os.File, intervals []Interval, large, small int64) ([]byte, error) {
+	var data []byte
+	for _, interval := range intervals {
+		shardId, shardOffset := interval.ToShardIdAndOffset(large, small)
+		if int(shardId) >= len(ecFiles) {
+			return nil, fmt.Errorf("shard %d out of range (have %d files)", shardId, len(ecFiles))
+		}
+		stat, err := ecFiles[shardId].Stat()
+		if err != nil {
+			return nil, fmt.Errorf("stat shard %d: %v", shardId, err)
+		}
+		if shardOffset+int64(interval.Size) > stat.Size() {
+			return nil, fmt.Errorf("read past end of shard %d: offset %d + size %d > fileSize %d",
+				shardId, shardOffset, interval.Size, stat.Size())
+		}
+		chunk := make([]byte, interval.Size)
+		n, err := ecFiles[shardId].ReadAt(chunk, shardOffset)
+		if err != nil {
+			return nil, fmt.Errorf("read shard %d offset %d: %v", shardId, shardOffset, err)
+		}
+		if n != int(interval.Size) {
+			return nil, fmt.Errorf("short read shard %d: got %d want %d", shardId, n, interval.Size)
+		}
+		data = append(data, chunk...)
+	}
+	return data, nil
+}

weed/storage/erasure_coding/ec_locate.go

@@ -64,7 +64,7 @@ func moveToNextBlock(blockIndex int, isLargeBlock bool, nLargeBlockRows int64) (
 
 func locateOffset(largeBlockLength, smallBlockLength int64, shardDatSize int64, offset int64) (blockIndex int, isLargeBlock bool, nLargeBlockRows int64, innerBlockOffset int64) {
 	largeRowSize := largeBlockLength * DataShardsCount
-	nLargeBlockRows = (shardDatSize - 1) / largeBlockLength
+	nLargeBlockRows = shardDatSize / largeBlockLength
 
 	// if offset is within the large block area
 	if offset < nLargeBlockRows*largeRowSize {

weed/storage/erasure_coding/ec_roundtrip_test.go

@@ -0,0 +1,388 @@
+package erasure_coding
+
+import (
+	"bytes"
+	"crypto/rand"
+	"fmt"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+)
+
+// TestEcReadRoundTrip tests the EC encode→read cycle via LocateData for various
+// .dat file sizes, paying special attention to the large/small block boundary.
+//
+// The nLargeBlockRows calculation must correctly distinguish between large and small
+// blocks. A previous bug (issue #8947) caused an off-by-one error when
+// shardDatSize was an exact multiple of largeBlockSize, leading to data corruption.
+func TestEcReadRoundTrip(t *testing.T) {
+	const (
+		large = largeBlockSize // 10000
+		small = smallBlockSize // 100
+	)
+
+	largeRowSize := large * DataShardsCount // 100000
+	smallRowSize := small * DataShardsCount // 1000
+
+	testCases := []struct {
+		name    string
+		datSize int64
+	}{
+		// Exact multiples of largeRowSize — triggers the nLargeBlockRows off-by-one bug
+		{"1_large_row_exact", int64(largeRowSize)},
+		{"2_large_rows_exact", int64(2 * largeRowSize)},
+		{"3_large_rows_exact", int64(3 * largeRowSize)},
+
+		// Just over a large row boundary — has small blocks
+		{"1_large_row_plus_1", int64(largeRowSize + 1)},
+		{"2_large_rows_plus_small", int64(2*largeRowSize + smallRowSize)},
+		{"1_large_row_plus_half_small", int64(largeRowSize + smallRowSize/2)},
+
+		// Just under a large row boundary — all small blocks
+		{"just_under_1_large_row", int64(largeRowSize - 1)},
+		{"just_under_2_large_rows", int64(2*largeRowSize - 1)},
+
+		// Small data — no large blocks at all
+		{"small_only", int64(smallRowSize * 3)},
+		{"small_single_row", int64(smallRowSize)},
+
+		// Boundary with mixed large and small
+		{"boundary_spanning", int64(largeRowSize + smallRowSize*5 + 50)},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			testEcRead(t, large, small, tc.datSize)
+		})
+	}
+}
+
+// testEcRead creates a .dat file, EC-encodes it, then verifies LocateData-based reads
+// return correct data at positions throughout the file (especially near the large/small
+// block boundary).
+func testEcRead(t *testing.T, large, small, datSize int64) {
+	t.Helper()
+
+	dir := t.TempDir()
+	baseFileName := fmt.Sprintf("%s/rt_%d", dir, datSize)
+
+	// 1. Create a .dat file with deterministic random data
+	originalData := make([]byte, datSize)
+	_, err := rand.Read(originalData)
+	require.NoError(t, err, "generating random data")
+
+	err = os.WriteFile(baseFileName+".dat", originalData, 0644)
+	require.NoError(t, err, "writing .dat file")
+
+	ctx := NewDefaultECContext("", 0)
+
+	// 2. EC encode with test block sizes
+	err = generateEcFiles(baseFileName, int(small), large, small, ctx)
+	require.NoError(t, err, "EC encoding")
+
+	// 3. Open EC shard files for reading
+	ecFiles, err := openEcFiles(baseFileName, true, ctx)
+	require.NoError(t, err, "opening EC files")
+	defer closeEcFiles(ecFiles)
+
+	ecStat, err := ecFiles[0].Stat()
+	require.NoError(t, err)
+	shardFileSize := ecStat.Size()
+
+	// Compute shardDatSize as the production code does when datFileSize is known
+	shardDatSizeFromDat := datSize / int64(ctx.DataShards)
+
+	// 4. Verify EC reads at various positions
+	largeRowSize := large * DataShardsCount
+	encoderLargeRows := datSize / int64(largeRowSize)
+	boundaryOffset := encoderLargeRows * int64(largeRowSize)
+
+	readSize := types.Size(small / 2) // read half a small block
+	testOffsets := collectTestOffsets(datSize, int64(readSize), boundaryOffset, large, small)
+
+	for _, offset := range testOffsets {
+		// Test with shardDatSize from datFileSize (the production path with fix)
+		intervals := LocateData(large, small, shardDatSizeFromDat, offset, readSize)
+		ecData, err := assembleFromIntervals(ecFiles, intervals, large, small)
+		require.NoError(t, err, "reading EC data at offset %d (datFileSize path)", offset)
+
+		expected := originalData[offset : offset+int64(readSize)]
+		if !bytes.Equal(expected, ecData) {
+			t.Errorf("EC read mismatch at offset %d (datFileSize path, shardDatSize=%d, nLargeBlockRows=%d)",
+				offset, shardDatSizeFromDat, shardDatSizeFromDat/large)
+		}
+
+		// Test with shardDatSize from ecdFileSize-1 (the fallback path for old volumes)
+		intervalsFallback := LocateData(large, small, shardFileSize-1, offset, readSize)
+		ecDataFallback, err := assembleFromIntervals(ecFiles, intervalsFallback, large, small)
+		if err == nil && !bytes.Equal(expected, ecDataFallback) {
+			// The fallback path may fail for exact multiples — log as warning
+			t.Logf("WARN: EC read mismatch at offset %d (fallback path, shardFileSize=%d)",
+				offset, shardFileSize)
+		}
+	}
+}
+
+// locateOffsetBuggy reimplements locateOffset with the old buggy formula:
+//
+//	nLargeBlockRows = (shardDatSize - 1) / largeBlockLength
+//
+// This caused an off-by-one error when shardDatSize was an exact multiple of
+// largeBlockLength, miscounting the number of large block rows.
+func locateOffsetBuggy(largeBlockLength, smallBlockLength int64, shardDatSize int64, offset int64) (blockIndex int, isLargeBlock bool, nLargeBlockRows int64, innerBlockOffset int64) {
+	largeRowSize := largeBlockLength * DataShardsCount
+	nLargeBlockRows = (shardDatSize - 1) / largeBlockLength // THE BUG
+
+	if offset < nLargeBlockRows*largeRowSize {
+		isLargeBlock = true
+		blockIndex = int(offset / largeBlockLength)
+		innerBlockOffset = offset % largeBlockLength
+		return
+	}
+
+	isLargeBlock = false
+	offset -= nLargeBlockRows * largeRowSize
+	blockIndex = int(offset / smallBlockLength)
+	innerBlockOffset = offset % smallBlockLength
+	return
+}
+
+// locateDataBuggy is LocateData using the old buggy locateOffset.
+func locateDataBuggy(largeBlockLength, smallBlockLength int64, shardDatSize int64, offset int64, size types.Size) []Interval {
+	blockIndex, isLargeBlock, nLargeBlockRows, innerBlockOffset := locateOffsetBuggy(largeBlockLength, smallBlockLength, shardDatSize, offset)
+
+	var intervals []Interval
+	for size > 0 {
+		blockRemaining := largeBlockLength - innerBlockOffset
+		if !isLargeBlock {
+			blockRemaining = smallBlockLength - innerBlockOffset
+		}
+		if blockRemaining <= 0 {
+			blockIndex, isLargeBlock = moveToNextBlock(blockIndex, isLargeBlock, nLargeBlockRows)
+			innerBlockOffset = 0
+			continue
+		}
+		interval := Interval{
+			BlockIndex:          blockIndex,
+			InnerBlockOffset:    innerBlockOffset,
+			IsLargeBlock:        isLargeBlock,
+			LargeBlockRowsCount: int(nLargeBlockRows),
+		}
+		if int64(size) <= blockRemaining {
+			interval.Size = size
+			intervals = append(intervals, interval)
+			return intervals
+		}
+		interval.Size = types.Size(blockRemaining)
+		intervals = append(intervals, interval)
+		size -= interval.Size
+		blockIndex, isLargeBlock = moveToNextBlock(blockIndex, isLargeBlock, nLargeBlockRows)
+		innerBlockOffset = 0
+	}
+	return intervals
+}
+
+// TestEcOffByOneBug_Issue8947 directly demonstrates the off-by-one bug.
+//
+// It creates a .dat file whose size is an exact multiple of (largeBlockSize * DataShards),
+// EC-encodes it, then shows that:
+//   - The OLD buggy formula produces WRONG data (data corruption)
+//   - The FIXED formula produces CORRECT data
+func TestEcOffByOneBug_Issue8947(t *testing.T) {
+	const (
+		large = largeBlockSize // 10000
+		small = smallBlockSize // 100
+	)
+
+	// datSize is exactly 2 large rows — each shard gets exactly 2*largeBlockSize bytes.
+	// The encoder produces 2 large block rows and 0 small block rows.
+	datSize := int64(2 * large * DataShardsCount) // 200000
+
+	dir := t.TempDir()
+	baseFileName := fmt.Sprintf("%s/bug_%d", dir, datSize)
+
+	originalData := make([]byte, datSize)
+	_, err := rand.Read(originalData)
+	require.NoError(t, err)
+
+	err = os.WriteFile(baseFileName+".dat", originalData, 0644)
+	require.NoError(t, err)
+
+	ctx := NewDefaultECContext("", 0)
+	err = generateEcFiles(baseFileName, int(small), large, small, ctx)
+	require.NoError(t, err, "EC encoding")
+
+	ecFiles, err := openEcFiles(baseFileName, true, ctx)
+	require.NoError(t, err)
+	defer closeEcFiles(ecFiles)
+
+	// shardDatSize = datFileSize / DataShards = 2 * largeBlockSize
+	// This is an EXACT multiple of largeBlockSize.
+	shardDatSize := datSize / int64(ctx.DataShards) // = 2 * large = 20000
+
+	// The encoder used 2 large block rows, 0 small block rows.
+	// Correct: nLargeBlockRows = 20000 / 10000 = 2
+	// Buggy:   nLargeBlockRows = (20000 - 1) / 10000 = 1  ← OFF BY ONE
+	fixedRows := shardDatSize / large
+	buggyRows := (shardDatSize - 1) / large
+	assert.Equal(t, int64(2), fixedRows, "fixed formula should give 2 large block rows")
+	assert.Equal(t, int64(1), buggyRows, "buggy formula gives only 1 (the bug)")
+
+	// Test reading from the 2nd large block row (offsets 100000–199999).
+	// With the buggy formula (nLargeBlockRows=1), this region is incorrectly
+	// treated as small blocks, causing reads from the WRONG shard positions.
+	readSize := types.Size(small / 2)
+
+	// Pick an offset well into the 2nd large block row so that the buggy formula
+	// computes a different (shard, offset) than the correct formula.
+	// At the very start of the 2nd row, both formulas coincidentally hit the same
+	// shard position. But further in, the small-block vs large-block addressing diverges.
+	offset := int64(large*DataShardsCount) + large + 50 // 110050: in 2nd large row, shard 1
+
+	// --- Fixed formula: reads correct data ---
+	fixedIntervals := LocateData(large, small, shardDatSize, offset, readSize)
+	fixedData, err := assembleFromIntervals(ecFiles, fixedIntervals, large, small)
+	require.NoError(t, err, "fixed LocateData read")
+
+	expected := originalData[offset : offset+int64(readSize)]
+	assert.True(t, bytes.Equal(expected, fixedData),
+		"FIXED formula should read correct data from 2nd large block row")
+
+	// --- Buggy formula: reads WRONG data ---
+	buggyIntervals := locateDataBuggy(large, small, shardDatSize, offset, readSize)
+	buggyData, err := assembleFromIntervals(ecFiles, buggyIntervals, large, small)
+	// The buggy formula might read from wrong offsets (possibly out of bounds),
+	// so an error is also evidence of the bug.
+	if err != nil {
+		t.Logf("Buggy formula caused read error (expected): %v", err)
+	} else {
+		assert.False(t, bytes.Equal(expected, buggyData),
+			"BUGGY formula should return WRONG data from 2nd large block row (demonstrating the corruption)")
+		n := 8
+		if len(expected) < n {
+			n = len(expected)
+		}
+		t.Logf("Buggy formula returned wrong data: expected first bytes %x, got %x",
+			expected[:n], buggyData[:n])
+	}
+
+	// Verify the bug mechanism: buggy formula misclassifies the 2nd large row as small blocks
+	assert.True(t, fixedIntervals[0].IsLargeBlock,
+		"fixed: offset %d should be in large blocks", offset)
+	assert.False(t, buggyIntervals[0].IsLargeBlock,
+		"buggy: offset %d is incorrectly classified as small blocks (the bug)", offset)
+
+	t.Logf("Fixed: nLargeBlockRows=%d, interval=%+v", fixedRows, fixedIntervals[0])
+	t.Logf("Buggy: nLargeBlockRows=%d, interval=%+v", buggyRows, buggyIntervals[0])
+}
+
+// TestEcDecodeDatRoundTrip tests the full WriteDatFile decode path using the production
+// block sizes with a small .dat file that fits within the small block region.
+func TestEcDecodeDatRoundTrip(t *testing.T) {
+	// With production sizes, datFileSize must be < DataShardsCount * ErasureCodingLargeBlockSize (10GB)
+	// to avoid needing huge test files. We test small block decode only.
+	// Each shard gets datSize/DataShards bytes in small 1MB blocks.
+	datSizes := []int64{
+		1000,                // tiny
+		int64(DataShardsCount) * ErasureCodingSmallBlockSize,     // exactly 1 small row (10MB)
+		int64(DataShardsCount)*ErasureCodingSmallBlockSize + 500, // 1 small row + partial
+	}
+
+	for _, datSize := range datSizes {
+		t.Run(fmt.Sprintf("size_%d", datSize), func(t *testing.T) {
+			testDecodeDat(t, datSize)
+		})
+	}
+}
+
+func testDecodeDat(t *testing.T, datSize int64) {
+	t.Helper()
+
+	dir := t.TempDir()
+	baseFileName := fmt.Sprintf("%s/dec_%d", dir, datSize)
+
+	// 1. Create .dat with random data
+	originalData := make([]byte, datSize)
+	_, err := rand.Read(originalData)
+	require.NoError(t, err)
+
+	err = os.WriteFile(baseFileName+".dat", originalData, 0644)
+	require.NoError(t, err)
+
+	ctx := NewDefaultECContext("", 0)
+
+	// 2. EC encode with PRODUCTION block sizes
+	err = generateEcFiles(baseFileName, 256*1024, ErasureCodingLargeBlockSize, ErasureCodingSmallBlockSize, ctx)
+	require.NoError(t, err, "EC encoding")
+
+	// 3. Decode via WriteDatFile
+	decodedBase := baseFileName + "_decoded"
+	shardFileNames := make([]string, DataShardsCount)
+	for i := 0; i < DataShardsCount; i++ {
+		shardFileNames[i] = fmt.Sprintf("%s%s", baseFileName, ctx.ToExt(i))
+	}
+
+	err = WriteDatFile(decodedBase, datSize, shardFileNames)
+	require.NoError(t, err, "WriteDatFile")
+
+	// 4. Verify decoded .dat matches original
+	decodedData, err := os.ReadFile(decodedBase + ".dat")
+	require.NoError(t, err)
+
+	assert.Equal(t, len(originalData), len(decodedData), "decoded .dat size mismatch")
+	if !bytes.Equal(originalData, decodedData) {
+		for i := 0; i < len(originalData) && i < len(decodedData); i++ {
+			if originalData[i] != decodedData[i] {
+				t.Fatalf("decoded .dat mismatch at byte %d (datSize=%d)", i, datSize)
+			}
+		}
+	}
+}
+
+// collectTestOffsets generates offsets to test, focusing on the large/small block boundary.
+func collectTestOffsets(datSize, readSize, boundaryOffset, large, small int64) []int64 {
+	offsets := []int64{0}
+
+	if datSize > readSize {
+		offsets = append(offsets, datSize/2)
+	}
+
+	// Near the large/small block boundary
+	if boundaryOffset > 0 && boundaryOffset < datSize {
+		for _, delta := range []int64{-large, -small, -1, 0, 1, small, large} {
+			off := boundaryOffset + delta
+			if off >= 0 && off+readSize <= datSize {
+				offsets = append(offsets, off)
+			}
+		}
+	}
+
+	// Near end of file
+	if datSize > readSize {
+		offsets = append(offsets, datSize-readSize)
+	}
+
+	return offsets
+}
+
+// assembleFromIntervals reads data from EC shard files according to the given intervals.
+func assembleFromIntervals(ecFiles []*os.File, intervals []Interval, large, small int64) ([]byte, error) {
+	var data []byte
+	for _, interval := range intervals {
+		shardId, shardOffset := interval.ToShardIdAndOffset(large, small)
+		chunk := make([]byte, interval.Size)
+		n, err := ecFiles[shardId].ReadAt(chunk, shardOffset)
+		if err != nil {
+			return nil, fmt.Errorf("read shard %d offset %d size %d: %v", shardId, shardOffset, interval.Size, err)
+		}
+		if n != int(interval.Size) {
+			return nil, fmt.Errorf("short read from shard %d: got %d, want %d", shardId, n, interval.Size)
+		}
+		data = append(data, chunk...)
+	}
+	return data, nil
+}

weed/storage/erasure_coding/ec_test.go

@@ -217,7 +217,9 @@ func (this Interval) sameAs(that Interval) bool {
 }
 
 func TestLocateData2(t *testing.T) {
-	intervals := LocateData(ErasureCodingLargeBlockSize, ErasureCodingSmallBlockSize, 3221225472, 21479557912, 4194339)
+	// Use ecdFileSize-1 to simulate the fallback path in LocateEcShardNeedleInterval
+	// when datFileSize is not available (old EC volumes without .vif datFileSize).
+	intervals := LocateData(ErasureCodingLargeBlockSize, ErasureCodingSmallBlockSize, 3221225472-1, 21479557912, 4194339)
 	assert.Equal(t, intervals, []Interval{
 		{BlockIndex: 4, InnerBlockOffset: 527128, Size: 521448, IsLargeBlock: false, LargeBlockRowsCount: 2},
 		{BlockIndex: 5, InnerBlockOffset: 0, Size: 1048576, IsLargeBlock: false, LargeBlockRowsCount: 2},
@@ -228,7 +230,8 @@ func TestLocateData2(t *testing.T) {
 }
 
 func TestLocateData3(t *testing.T) {
-	intervals := LocateData(ErasureCodingLargeBlockSize, ErasureCodingSmallBlockSize, 3221225472, 30782909808, 112568)
+	// Use ecdFileSize-1 to simulate the fallback path in LocateEcShardNeedleInterval
+	intervals := LocateData(ErasureCodingLargeBlockSize, ErasureCodingSmallBlockSize, 3221225472-1, 30782909808, 112568)
 	for _, interval := range intervals {
 		fmt.Printf("%+v\n", interval)
 	}
@@ -237,13 +240,29 @@ func TestLocateData3(t *testing.T) {
 	})
 }
 
+func TestLocateData_ExactMultiple_Issue8947(t *testing.T) {
+	// When datFileSize is available, shardDatSize = datFileSize / DataShards.
+	// For a 30GB volume with 10 data shards, shardDatSize = 3GB = 3 * ErasureCodingLargeBlockSize.
+	// The encoder produces 3 large block rows, 0 small block rows.
+	// nLargeBlockRows must be 3, not 2.
+	shardDatSize := int64(3) * ErasureCodingLargeBlockSize // 3GB per shard from datFileSize/DataShards
+
+	// Reading from the 3rd large block row (offsets 20GB-30GB) should work
+	offset := int64(2) * ErasureCodingLargeBlockSize * DataShardsCount // 20GB
+	intervals := LocateData(ErasureCodingLargeBlockSize, ErasureCodingSmallBlockSize, shardDatSize, offset, 1024)
+	assert.Equal(t, 1, len(intervals))
+	assert.True(t, intervals[0].IsLargeBlock, "data in 3rd large row should be in large blocks")
+	assert.Equal(t, 3, intervals[0].LargeBlockRowsCount)
+	assert.Equal(t, 20, intervals[0].BlockIndex) // block 20 = shard 0 of 3rd row
+}
+
 func TestLocateData_Issue8179(t *testing.T) {
 	large := int64(10000)
 	small := int64(100)
 	shardSize := int64(259092) // Resulting in nLargeBlockRows = 25 as seen in panic log
 
 	// Testing range through the large-to-small transition boundary
-	nLargeBlockRows := (shardSize - 1) / large
+	nLargeBlockRows := shardSize / large
 	largeAreaSize := nLargeBlockRows * int64(DataShardsCount) * large
 
 	for offset := largeAreaSize - 500; offset < largeAreaSize+500; offset++ {

weed/storage/erasure_coding/ec_volume.go

@@ -294,14 +294,17 @@ func (ev *EcVolume) LocateEcShardNeedle(needleId types.NeedleId, version needle.
 
 func (ev *EcVolume) LocateEcShardNeedleInterval(version needle.Version, offset int64, size types.Size) (intervals []Interval) {
 	shard := ev.Shards[0]
-	// Usually shard will be padded to round of ErasureCodingSmallBlockSize.
-	// So in most cases, if shardSize equals to n * ErasureCodingLargeBlockSize,
-	// the data would be in small blocks.
-	shardSize := shard.ecdFileSize - 1
+	var shardSize int64
 	if ev.datFileSize > 0 {
-		// To get the correct LargeBlockRowsCount
-		// use datFileSize to calculate the shardSize to match the EC encoding logic.
+		// Use datFileSize to calculate the shardSize to match the EC encoding logic.
+		// This is the authoritative value stored in .vif during EC encoding.
 		shardSize = ev.datFileSize / int64(ev.ECContext.DataShards)
+	} else {
+		// Fallback for old EC volumes without datFileSize in .vif.
+		// Subtract 1 to handle the ambiguous case where ecdFileSize is an exact
+		// multiple of ErasureCodingLargeBlockSize but the data is actually in small
+		// blocks (e.g., datFileSize was just under DataShards*ErasureCodingLargeBlockSize).
+		shardSize = shard.ecdFileSize - 1
 	}
 	// calculate the locations in the ec shards
 	intervals = LocateData(ErasureCodingLargeBlockSize, ErasureCodingSmallBlockSize, shardSize, offset, types.Size(needle.GetActualSize(size, version)))

weed/storage/needle/needle_read.go

@@ -55,7 +55,6 @@ func (n *Needle) ReadBytes(bytes []byte, offset int64, size Size, version Versio
 	if n.Size != size {
 		if OffsetSize == 4 && offset < int64(MaxPossibleVolumeSize) {
 			stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorSizeMismatchOffsetSize).Inc()
-			glog.Errorf("entry not found: offset %d found id %x size %d, expected size %d", offset, n.Id, n.Size, size)
 			return ErrorSizeMismatch
 		}
 		stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorSizeMismatch).Inc()

weed/storage/store_ec.go

@@ -169,7 +169,7 @@ func (s *Store) ReadEcShardNeedle(vid needle.VolumeId, n *needle.Needle, onReadS
 
 			err = n.ReadBytes(bytes, offset.ToActualOffset(), size, localEcVolume.Version)
 			if err != nil {
-				return 0, fmt.Errorf("readbytes: %w", err)
+				return 0, fmt.Errorf("ec volume %d needle %s offset %d size %d: %w", vid, n.String(), offset.ToActualOffset(), size, err)
 			}
 
 			return len(bytes), nil

weed/util/http/client/http_client.go

@@ -195,6 +195,66 @@ func getClientCaCert(clientName ClientName) ([]byte, string, error) {
 	return getFileContentFromSecurityConfiguration(clientName, "ca")
 }
 
+// NewHttpClientWithTLS creates an HTTPClient with explicit TLS certificate
+// parameters instead of reading from the global security configuration.
+// This is used by filer.sync to create per-cluster HTTP clients when clusters
+// use different certificates.
+func NewHttpClientWithTLS(certFile, keyFile, caFile string, insecureSkipVerify bool, opts ...HttpClientOpt) (*HTTPClient, error) {
+	httpClient := HTTPClient{}
+	httpClient.expectHttpsScheme = true
+	var tlsConfig *tls.Config
+
+	if (certFile == "") != (keyFile == "") {
+		return nil, fmt.Errorf("both cert and key are required for mTLS, got cert=%q key=%q", certFile, keyFile)
+	}
+
+	var clientCert *tls.Certificate
+	if certFile != "" && keyFile != "" {
+		cert, err := tls.LoadX509KeyPair(certFile, keyFile)
+		if err != nil {
+			return nil, fmt.Errorf("error loading client certificate and key: %s", err)
+		}
+		clientCert = &cert
+	}
+
+	var caCertPool *x509.CertPool
+	if caFile != "" {
+		caCert, err := os.ReadFile(caFile)
+		if err != nil {
+			return nil, fmt.Errorf("error reading CA cert %s: %s", caFile, err)
+		}
+		caCertPool, err = createHTTPClientCertPool(caCert, caFile)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if clientCert != nil || caCertPool != nil || insecureSkipVerify {
+		tlsConfig = &tls.Config{
+			Certificates:       []tls.Certificate{},
+			RootCAs:            caCertPool,
+			InsecureSkipVerify: insecureSkipVerify,
+		}
+		if clientCert != nil {
+			tlsConfig.Certificates = append(tlsConfig.Certificates, *clientCert)
+		}
+	}
+
+	httpClient.Transport = &http.Transport{
+		MaxIdleConns:        1024,
+		MaxIdleConnsPerHost: 1024,
+		TLSClientConfig:     tlsConfig,
+	}
+	httpClient.Client = &http.Client{
+		Transport: httpClient.Transport,
+	}
+
+	for _, opt := range opts {
+		opt(&httpClient)
+	}
+	return &httpClient, nil
+}
+
 func createHTTPClientCertPool(certContent []byte, fileName string) (*x509.CertPool, error) {
 	certPool := x509.NewCertPool()
 	if len(certContent) == 0 {

weed/util/http/http_global_client_util.go

@@ -21,6 +21,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 
 	"github.com/seaweedfs/seaweedfs/weed/security"
+	util_http_client "github.com/seaweedfs/seaweedfs/weed/util/http/client"
 )
 
 var ErrNotFound = fmt.Errorf("not found")
@@ -202,6 +203,16 @@ func GetUrlStream(url string, values url.Values, readFn func(io.Reader) error) e
 }
 
 func DownloadFile(fileUrl string, jwt string, offset ...int64) (filename string, header http.Header, resp *http.Response, e error) {
+	return DownloadFileWithClient(GetGlobalHttpClient(), fileUrl, jwt, offset...)
+}
+
+// DownloadFileWithClient is like DownloadFile but uses the provided HTTP client
+// instead of the global one. This is used by filer.sync to download from
+// remote clusters that use different TLS certificates.
+func DownloadFileWithClient(client *util_http_client.HTTPClient, fileUrl string, jwt string, offset ...int64) (filename string, header http.Header, resp *http.Response, e error) {
+	if client == nil {
+		return "", nil, nil, fmt.Errorf("nil HTTP client in DownloadFileWithClient")
+	}
 	req, err := http.NewRequest(http.MethodGet, fileUrl, nil)
 	if err != nil {
 		return "", nil, nil, err
@@ -217,7 +228,7 @@ func DownloadFile(fileUrl string, jwt string, offset ...int64) (filename string,
 		req.Header.Set("Range", fmt.Sprintf("bytes=%d-", rangeOffset))
 	}
 
-	response, err := GetGlobalHttpClient().Do(req)
+	response, err := client.Do(req)
 	if err != nil {
 		return "", nil, nil, err
 	}

weed/util/version/constants.go

@@ -9,7 +9,7 @@ import (
 
 var (
 	MAJOR_VERSION  = int32(4)
-	MINOR_VERSION  = int32(18)
+	MINOR_VERSION  = int32(19)
 	VERSION_NUMBER = fmt.Sprintf("%d.%02d", MAJOR_VERSION, MINOR_VERSION)
 	VERSION        = util.SizeLimit + " " + VERSION_NUMBER
 	COMMIT         = ""

weed/worker/tasks/erasure_coding/ec_task.go

@@ -15,7 +15,9 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
+	"github.com/seaweedfs/seaweedfs/weed/storage/idx"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	storagetypes "github.com/seaweedfs/seaweedfs/weed/storage/types"
 	"github.com/seaweedfs/seaweedfs/weed/storage/volume_info"
 	"github.com/seaweedfs/seaweedfs/weed/worker/types"
 	"github.com/seaweedfs/seaweedfs/weed/worker/types/base"
@@ -150,10 +152,16 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP
 	}
 
 	// Step 2: Copy volume files to worker
+	// The .idx and .dat are copied as separate network transfers, with .idx
+	// copied first. If a write lands on the source after the .idx copy, the
+	// .dat will include extra data not referenced by .idx (harmless).
+	// verifyDatIdxConsistency() in generateEcShardsLocally catches the reverse
+	// case where .idx references data past .dat.
 	t.ReportProgressWithStage(25.0, "Copying volume files to worker")
 	t.GetLogger().Info("Copying volume files to worker")
 	localFiles, err := t.copyVolumeFilesToWorker(ctx, taskWorkDir)
 	if err != nil {
+		t.rollbackReadonly(ctx)
 		return fmt.Errorf("failed to copy volume files: %v", err)
 	}
 
@@ -162,6 +170,7 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP
 	t.GetLogger().Info("Generating EC shards locally")
 	shardFiles, err := t.generateEcShardsLocally(localFiles, taskWorkDir)
 	if err != nil {
+		t.rollbackReadonly(ctx)
 		return fmt.Errorf("failed to generate EC shards: %v", err)
 	}
 
@@ -259,7 +268,35 @@ func (t *ErasureCodingTask) markVolumeReadonly(ctx context.Context) error {
 		})
 }
 
-// copyVolumeFilesToWorker copies .dat and .idx files from source server to local worker
+// rollbackReadonly is a best-effort rollback of markVolumeReadonly, used when the
+// EC task fails before any shards are distributed. Logs but does not return errors.
+// Uses a fresh context with timeout since the caller's ctx may already be cancelled.
+func (t *ErasureCodingTask) rollbackReadonly(_ context.Context) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	if err := t.markVolumeWritable(ctx); err != nil {
+		glog.Warningf("failed to restore volume %d to writable after EC task failure: %v", t.volumeID, err)
+	} else {
+		glog.V(0).Infof("restored volume %d to writable after EC task failure", t.volumeID)
+	}
+}
+
+// markVolumeWritable restores the volume to writable on the source server.
+func (t *ErasureCodingTask) markVolumeWritable(ctx context.Context) error {
+	return operation.WithVolumeServerClient(false, pb.ServerAddress(t.server), t.grpcDialOption,
+		func(client volume_server_pb.VolumeServerClient) error {
+			_, err := client.VolumeMarkWritable(ctx, &volume_server_pb.VolumeMarkWritableRequest{
+				VolumeId: t.volumeID,
+			})
+			return err
+		})
+}
+
+// copyVolumeFilesToWorker copies .idx and .dat files from source server to local worker.
+// The .idx is copied first, then .dat. Both copies are capped to the sizes reported by
+// ReadVolumeFileStatus. If a write lands after .idx is copied, .dat may include extra
+// data not referenced by .idx (harmless). The reverse (idx referencing data past .dat)
+// is caught by verifyDatIdxConsistency in generateEcShardsLocally.
 func (t *ErasureCodingTask) copyVolumeFilesToWorker(ctx context.Context, workDir string) (map[string]string, error) {
 	localFiles := make(map[string]string)
 
@@ -277,31 +314,14 @@ func (t *ErasureCodingTask) copyVolumeFilesToWorker(ctx context.Context, workDir
 		"idx_file_size_bytes": fileStatus.GetIdxFileSize(),
 	}).Info("Starting volume file copy from source server")
 
-	// Copy .dat file
-	datFile := filepath.Join(workDir, fmt.Sprintf("%d.dat", t.volumeID))
-	if err := t.copyFileFromSource(ctx, ".dat", datFile, fileStatus.GetCompactionRevision(), fileStatus.GetDatFileSize()); err != nil {
-		return nil, fmt.Errorf("failed to copy .dat file: %v", err)
-	}
-	localFiles["dat"] = datFile
-
-	// Log .dat file size
-	if info, err := os.Stat(datFile); err == nil {
-		t.GetLogger().WithFields(map[string]interface{}{
-			"file_type":  ".dat",
-			"file_path":  datFile,
-			"size_bytes": info.Size(),
-			"size_mb":    float64(info.Size()) / (1024 * 1024),
-		}).Info("Volume data file copied successfully")
-	}
-
-	// Copy .idx file
+	// Copy .idx file FIRST — if a write lands on the source after this copy,
+	// the .dat copy will include the new data but .idx won't reference it.
 	idxFile := filepath.Join(workDir, fmt.Sprintf("%d.idx", t.volumeID))
 	if err := t.copyFileFromSource(ctx, ".idx", idxFile, fileStatus.GetCompactionRevision(), fileStatus.GetIdxFileSize()); err != nil {
 		return nil, fmt.Errorf("failed to copy .idx file: %v", err)
 	}
 	localFiles["idx"] = idxFile
 
-	// Log .idx file size
 	if info, err := os.Stat(idxFile); err == nil {
 		t.GetLogger().WithFields(map[string]interface{}{
 			"file_type":  ".idx",
@@ -311,6 +331,22 @@ func (t *ErasureCodingTask) copyVolumeFilesToWorker(ctx context.Context, workDir
 		}).Info("Volume index file copied successfully")
 	}
 
+	// Copy .dat file SECOND — guaranteed to have at least as much data as .idx references.
+	datFile := filepath.Join(workDir, fmt.Sprintf("%d.dat", t.volumeID))
+	if err := t.copyFileFromSource(ctx, ".dat", datFile, fileStatus.GetCompactionRevision(), fileStatus.GetDatFileSize()); err != nil {
+		return nil, fmt.Errorf("failed to copy .dat file: %v", err)
+	}
+	localFiles["dat"] = datFile
+
+	if info, err := os.Stat(datFile); err == nil {
+		t.GetLogger().WithFields(map[string]interface{}{
+			"file_type":  ".dat",
+			"file_path":  datFile,
+			"size_bytes": info.Size(),
+			"size_mb":    float64(info.Size()) / (1024 * 1024),
+		}).Info("Volume data file copied successfully")
+	}
+
 	return localFiles, nil
 }
 
@@ -401,16 +437,23 @@ func (t *ErasureCodingTask) generateEcShardsLocally(localFiles map[string]string
 
 	glog.V(1).Infof("Generating EC shards from local files: dat=%s, idx=%s", datFile, idxFile)
 
+	// Verify .dat and .idx are consistent before EC encoding.
+	// Since they were copied as separate network transfers, the .idx may have
+	// entries pointing past the end of .dat if a write landed between the copies.
+	if err := verifyDatIdxConsistency(datFile, idxFile); err != nil {
+		return nil, fmt.Errorf("dat/idx consistency check failed: %v", err)
+	}
+
+	// Generate .ecx file from .idx BEFORE EC shards to prevent inconsistency.
+	if err := erasure_coding.WriteSortedFileFromIdx(baseName, ".ecx"); err != nil {
+		return nil, fmt.Errorf("failed to generate .ecx file: %v", err)
+	}
+
 	// Generate EC shard files (.ec00 ~ .ec13)
 	if err := erasure_coding.WriteEcFiles(baseName); err != nil {
 		return nil, fmt.Errorf("failed to generate EC shard files: %v", err)
 	}
 
-	// Generate .ecx file from .idx (use baseName, not full idx path)
-	if err := erasure_coding.WriteSortedFileFromIdx(baseName, ".ecx"); err != nil {
-		return nil, fmt.Errorf("failed to generate .ecx file: %v", err)
-	}
-
 	// Collect generated shard file paths and log details
 	var generatedShards []string
 	var totalShardSize int64
@@ -588,3 +631,62 @@ func (t *ErasureCodingTask) getReplicas() []string {
 	}
 	return replicas
 }
+
+// verifyDatIdxConsistency checks that all .idx entries reference data within the
+// .dat file. Since .dat and .idx are copied as separate network transfers, the
+// .idx may have entries from writes that landed after the .dat was copied.
+func verifyDatIdxConsistency(datFile, idxFile string) error {
+	datInfo, err := os.Stat(datFile)
+	if err != nil {
+		return fmt.Errorf("stat dat file: %v", err)
+	}
+	datSize := datInfo.Size()
+
+	// Read volume version from superblock to compute actual needle sizes
+	df, err := os.Open(datFile)
+	if err != nil {
+		return fmt.Errorf("open dat file: %v", err)
+	}
+	defer df.Close()
+
+	versionBytes := make([]byte, 1)
+	if _, err := df.ReadAt(versionBytes, 0); err != nil {
+		return fmt.Errorf("read version byte: %v", err)
+	}
+	version := needle.Version(versionBytes[0])
+
+	idxF, err := os.Open(idxFile)
+	if err != nil {
+		return fmt.Errorf("open idx file: %v", err)
+	}
+	defer idxF.Close()
+
+	var maxEnd int64
+	var maxEndNeedleId storagetypes.NeedleId
+	var entryCount int64
+	err = idx.WalkIndexFile(idxF, 0, func(key storagetypes.NeedleId, offset storagetypes.Offset, size storagetypes.Size) error {
+		entryCount++
+		if size.IsDeleted() {
+			return nil
+		}
+		end := offset.ToActualOffset() + needle.GetActualSize(size, version)
+		if end > maxEnd {
+			maxEnd = end
+			maxEndNeedleId = key
+		}
+		return nil
+	})
+	if err != nil {
+		return fmt.Errorf("walk idx file: %v", err)
+	}
+
+	if maxEnd > datSize {
+		return fmt.Errorf(
+			"idx references data beyond dat file: needle %d ends at offset %d but dat file is only %d bytes (%d entries total)",
+			maxEndNeedleId, maxEnd, datSize, entryCount,
+		)
+	}
+
+	glog.V(1).Infof("dat/idx consistency check passed: %d entries, max offset %d, dat size %d", entryCount, maxEnd, datSize)
+	return nil
+}