Merge remote-tracking branch 'origin/master' into nfs-enable-inode-index

k8s/charts/seaweedfs/Chart.yaml

@@ -1,6 +1,6 @@
 apiVersion: v1
 description: SeaweedFS
 name: seaweedfs
-appVersion: "4.49"
+appVersion: "4.50"
 # Dev note: Trigger a helm chart release by `git tag -a helm-<version>`
-version: 4.49.0
+version: 4.50.0

k8s/charts/seaweedfs/templates/filer/filer-statefulset.yaml

@@ -170,9 +170,6 @@ spec:
               {{- if .Values.filer.metricsIp }}
               -metricsIp={{ .Values.filer.metricsIp }} \
               {{- end }}
-              {{- if .Values.filer.redirectOnRead }}
-              -redirectOnRead \
-              {{- end }}
               {{- if .Values.filer.disableHttp }}
               -disableHttp \
               {{- end }}

k8s/charts/seaweedfs/values.yaml

@@ -740,9 +740,6 @@ filer:
   # encrypt data on volume servers
   encryptVolumeData: false
 
-  # Whether proxy or redirect to volume server during file GET request
-  redirectOnRead: false
-
   # Limit sub dir listing size (default 100000)
   dirListLimit: 100000
 

seaweed-volume/src/server/handlers.rs

@@ -3039,50 +3039,62 @@ pub async fn post_handler(
     let needle_etag = n.etag();
     let needle_has_name = n.has_name();
 
-    let write_result = write_upload_needle(&state, vid, n, fsync).await;
-
     // Replicate to remote volume servers if this volume has replicas.
     // Matches Go's GetWritableRemoteReplications: skip if copy_count == 1.
+    let needs_replication = !is_replicate && {
+        let store = state.store.read().unwrap();
+        store
+            .find_volume(vid)
+            .is_some_and(|(_, v)| v.super_block().replica_placement.get_copy_count() > 1)
+    };
+
+    // The fan-out runs concurrently with the local needle write, like Go's
+    // ReplicatedWrite: it reads the already-buffered request body, so the two
+    // sides share no mutable state. The ack still waits for both.
+    let replication = if needs_replication {
+        let state_clone = state.clone();
+        let path_clone = path.clone();
+        let query_clone = query.clone();
+        let headers_clone = headers.clone();
+        let body_clone = body.clone();
+        Some(tokio::spawn(async move {
+            do_replicated_request(
+                &state_clone,
+                vid.0,
+                Method::POST,
+                &path_clone,
+                &query_clone,
+                &headers_clone,
+                Some(body_clone),
+            )
+            .await
+        }))
+    } else {
+        None
+    };
+
+    let write_result = write_upload_needle(&state, vid, n, fsync).await;
+
     // No leader-known gate: when the lookup cannot run, the write must fail
-    // rather than be acknowledged on fewer copies than required.
-    if !is_replicate && write_result.is_ok() {
-        let needs_replication = {
-            let store = state.store.read().unwrap();
-            store
-                .find_volume(vid)
-                .is_some_and(|(_, v)| v.super_block().replica_placement.get_copy_count() > 1)
-        };
-        if needs_replication {
-            let state_clone = state.clone();
-            let path_clone = path.clone();
-            let query_clone = query.clone();
-            let headers_clone = headers.clone();
-            let body_clone = body.clone();
-            let replication = tokio::spawn(async move {
-                do_replicated_request(
-                    &state_clone,
-                    vid.0,
-                    Method::POST,
-                    &path_clone,
-                    &query_clone,
-                    &headers_clone,
-                    Some(body_clone),
-                )
-                .await
-            });
-            let replication_result = replication
-                .await
+    // rather than be acknowledged on fewer copies than required. The local
+    // write result takes precedence, matching Go's error ordering.
+    let replication_result = match replication {
+        Some(task) => Some(
+            task.await
                 .map_err(|e| format!("replication task failed: {}", e))
-                .flatten();
-            if let Err(e) = replication_result {
-                tracing::error!("replicated write failed: {}", e);
-                return json_error_with_query(
-                    StatusCode::INTERNAL_SERVER_ERROR,
-                    format!("replication failed: {}", e),
-                    Some(&query),
-                );
-            }
-        }
+                .flatten(),
+        ),
+        None => None,
+    };
+    if write_result.is_ok()
+        && let Some(Err(e)) = replication_result
+    {
+        tracing::error!("replicated write failed: {}", e);
+        return json_error_with_query(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            format!("replication failed: {}", e),
+            Some(&query),
+        );
     }
 
     let resp = match write_result {

test/s3tables/catalog_trino/trino_catalog_test.go

@@ -496,9 +496,7 @@ func waitForTrino(t *testing.T, containerName string, timeout time.Duration) {
 			return
 		} else {
 			lastOutput = output
-			outputStr := string(output)
-			if strings.Contains(outputStr, "No such container") ||
-				strings.Contains(outputStr, "is not running") {
+			if trinoContainerExited(output) {
 				containerDead = true
 				break
 			}
@@ -507,18 +505,25 @@ func waitForTrino(t *testing.T, containerName string, timeout time.Duration) {
 	}
 
 	// If we can't connect to system catalog, try to at least connect to Trino server
-	cmd := exec.Command("docker", "exec", containerName, "trino", "--version")
-	if err := cmd.Run(); err == nil {
+	if output, err := exec.Command("docker", "exec", containerName, "trino", "--version").CombinedOutput(); err == nil {
 		// Trino process is running, even if catalog isn't ready yet
 		// Give it a bit more time
 		time.Sleep(5 * time.Second)
 		return
+	} else {
+		lastOutput = output
+		containerDead = containerDead || trinoContainerExited(output)
 	}
 
 	if containerDead {
 		t.Fatalf("Trino container %s exited before becoming ready\nLast output:\n%s\n%s", containerName, string(lastOutput), trinoContainerDiagnostics(containerName))
 	}
-	t.Fatalf("Timed out waiting for Trino to be ready\nLast output:\n%s", string(lastOutput))
+	t.Fatalf("Timed out waiting for Trino to be ready\nLast output:\n%s\n%s", string(lastOutput), trinoContainerDiagnostics(containerName))
+}
+
+func trinoContainerExited(output []byte) bool {
+	s := string(output)
+	return strings.Contains(s, "No such container") || strings.Contains(s, "is not running")
 }
 
 func trinoContainerDiagnostics(containerName string) string {

test/volume_server/http/read_path_variants_test.go

@@ -170,11 +170,11 @@ func TestConditionalHeaderPrecedenceAndInvalidIfModifiedSince(t *testing.T) {
 	precedenceReq.Header.Set("If-None-Match", "\"definitely-different-etag\"")
 	precedenceResp := framework.DoRequest(t, client, precedenceReq)
 	precedenceBody := framework.ReadAllAndClose(t, precedenceResp)
-	if precedenceResp.StatusCode != http.StatusNotModified {
-		t.Fatalf("conditional precedence expected 304, got %d", precedenceResp.StatusCode)
+	if precedenceResp.StatusCode != http.StatusOK {
+		t.Fatalf("mismatched If-None-Match overrides If-Modified-Since: expected 200, got %d", precedenceResp.StatusCode)
 	}
-	if len(precedenceBody) != 0 {
-		t.Fatalf("conditional precedence expected empty body, got %d bytes", len(precedenceBody))
+	if string(precedenceBody) != string(payload) {
+		t.Fatalf("mismatched If-None-Match body mismatch: got %q want %q", string(precedenceBody), string(payload))
 	}
 
 	invalidIMSReq := mustNewRequest(t, http.MethodGet, clusterHarness.VolumeAdminURL()+"/"+fid)

weed/filer/etcd/etcd_store.go

@@ -181,7 +181,7 @@ func (store *EtcdStore) DeleteFolderChildren(ctx context.Context, fullpath weed_
 func (store *EtcdStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
 	directoryPrefix := genDirectoryKeyPrefix(dirPath, prefix)
 	lastFileStart := directoryPrefix
-	if startFileName != "" {
+	if startFileName > prefix {
 		lastFileStart = genDirectoryKeyPrefix(dirPath, startFileName)
 	}
 

weed/filer/filerstore_wrapper.go

@@ -438,46 +438,39 @@ func (fsw *FilerStoreWrapper) prefixFilterEntries(ctx context.Context, dirPath u
 		return actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, eachEntryFunc)
 	}
 
-	var notPrefixed []*Entry
-	lastFileName, err = actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *Entry) (bool, error) {
-		notPrefixed = append(notPrefixed, entry)
-		return true, nil
-	})
-	if err != nil {
-		return
-	}
-
+	// lastFileName must stay at the last name delivered to eachEntryFunc; the
+	// raw scan cursor runs ahead over non-matching names and cannot be returned.
 	count := int64(0)
-	for count < limit && len(notPrefixed) > 0 {
-		for _, entry := range notPrefixed {
-			if strings.HasPrefix(entry.Name(), prefix) {
-				count++
-				res, resErr := eachEntryFunc(entry)
-
-				if resErr != nil {
-					err = fmt.Errorf("failed to process eachEntryFunc for entry %q: %w", entry.Name(), resErr)
-					return
-				}
-
-				if !res {
-					return
-				}
-				if count >= limit {
-					break
-				}
-			}
+	scanName, scanInclusive := startFileName, includeStartFile
+	for count < limit {
+		var page []*Entry
+		if scanName, err = actualStore.ListDirectoryEntries(ctx, dirPath, scanName, scanInclusive, limit, func(entry *Entry) (bool, error) {
+			page = append(page, entry)
+			return true, nil
+		}); err != nil {
+			return
 		}
-		if count < limit && lastFileName < prefix {
-			notPrefixed = notPrefixed[:0]
-			lastFileName, err = actualStore.ListDirectoryEntries(ctx, dirPath, lastFileName, false, limit, func(entry *Entry) (bool, error) {
-				notPrefixed = append(notPrefixed, entry)
-				return true, nil
-			})
-			if err != nil {
+		scanInclusive = false
+		if len(page) == 0 {
+			return
+		}
+		for _, entry := range page {
+			if !strings.HasPrefix(entry.Name(), prefix) {
+				continue
+			}
+			count++
+			res, resErr := eachEntryFunc(entry)
+			if resErr != nil {
+				err = fmt.Errorf("failed to process eachEntryFunc for entry %q: %w", entry.Name(), resErr)
 				return
 			}
-		} else {
-			break
+			lastFileName = entry.Name()
+			if !res || count >= limit {
+				return
+			}
+		}
+		if scanName >= prefix && !strings.HasPrefix(scanName, prefix) {
+			return
 		}
 	}
 	return

weed/filer/filerstore_wrapper_test.go

@@ -205,3 +205,41 @@ func TestFilerStoreWrapperRollbackSucceedsWithCancelledContext(t *testing.T) {
 	assert.NoError(t, wrapper.RollbackTransaction(cancelledCtx()))
 	assert.NoError(t, wrapper.RollbackTransaction(expiredCtx()))
 }
+
+type noPrefixListStore struct {
+	*stubFilerStore
+}
+
+func (s *noPrefixListStore) ListDirectoryPrefixedEntries(context.Context, util.FullPath, string, bool, int64, string, ListEachEntryFunc) (string, error) {
+	return "", ErrUnsupportedListDirectoryPrefixed
+}
+
+// The wrapper fallback must return the last delivered name as the cursor, not
+// the end of the raw page, and must keep paging while prefix matches can still
+// follow the raw cursor.
+func TestPrefixFilterEntriesCursorStaysAtLastDelivered(t *testing.T) {
+	store := &noPrefixListStore{newStubFilerStore()}
+	wrapper := NewFilerStoreWrapper(store)
+	ctx := context.Background()
+	for _, name := range []string{"a", "b1", "b2", "b3", "b4"} {
+		require.NoError(t, store.InsertEntry(ctx, &Entry{FullPath: util.FullPath("/d/" + name)}))
+	}
+
+	var delivered []string
+	lastFileName, err := wrapper.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), "", false, 2, "b", func(entry *Entry) (bool, error) {
+		delivered = append(delivered, entry.Name())
+		return true, nil
+	})
+	require.NoError(t, err)
+	assert.Equal(t, []string{"b1", "b2"}, delivered)
+	assert.Equal(t, "b2", lastFileName)
+
+	delivered = delivered[:0]
+	lastFileName, err = wrapper.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), lastFileName, false, 10, "b", func(entry *Entry) (bool, error) {
+		delivered = append(delivered, entry.Name())
+		return true, nil
+	})
+	require.NoError(t, err)
+	assert.Equal(t, []string{"b3", "b4"}, delivered)
+	assert.Equal(t, "b4", lastFileName)
+}

weed/filer/hbase/hbase_store.go

@@ -156,7 +156,11 @@ func (store *HbaseStore) ListDirectoryEntries(ctx context.Context, dirPath util.
 func (store *HbaseStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
 	family := map[string][]string{store.cfMetaDir: {COLUMN_NAME}}
 	expectedPrefix := []byte(dirPath.Child(prefix))
-	scan, err := hrpc.NewScanRange(ctx, store.table, expectedPrefix, nil, hrpc.Families(family))
+	scanStart := expectedPrefix
+	if resumeKey := []byte(dirPath.Child(startFileName)); bytes.Compare(resumeKey, scanStart) > 0 {
+		scanStart = resumeKey
+	}
+	scan, err := hrpc.NewScanRange(ctx, store.table, scanStart, nil, hrpc.Families(family))
 	if err != nil {
 		return lastFileName, err
 	}
@@ -188,7 +192,7 @@ func (store *HbaseStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPa
 
 		value := cell.Value
 
-		if fileName == startFileName && !includeStartFile {
+		if fileName < startFileName || (fileName == startFileName && !includeStartFile) {
 			continue
 		}
 

weed/filer/leveldb/leveldb_store.go

@@ -221,7 +221,7 @@ func (store *LevelDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dir
 
 	directoryPrefix := genDirectoryKeyPrefix(dirPath, prefix)
 	lastFileStart := directoryPrefix
-	if startFileName != "" {
+	if startFileName > prefix {
 		lastFileStart = genDirectoryKeyPrefix(dirPath, startFileName)
 	}
 

weed/filer/leveldb/leveldb_store_test.go

@@ -110,3 +110,28 @@ func BenchmarkInsertEntry(b *testing.B) {
 		store.InsertEntry(ctx, entry)
 	}
 }
+
+func TestListDirectoryPrefixedEntries_StartBelowPrefix(t *testing.T) {
+	store := &LevelDBStore{}
+	store.initialize(t.TempDir())
+	defer store.Shutdown()
+
+	ctx := context.Background()
+	for _, name := range []string{"item-a", "item-a.versions", "item-b.versions"} {
+		if err := store.InsertEntry(ctx, &filer.Entry{FullPath: util.FullPath("/d/" + name)}); err != nil {
+			t.Fatalf("insert %s: %v", name, err)
+		}
+	}
+
+	var names []string
+	_, err := store.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), "item", true, 100, "item-b", func(entry *filer.Entry) (bool, error) {
+		names = append(names, entry.Name())
+		return true, nil
+	})
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	if len(names) != 1 || names[0] != "item-b.versions" {
+		t.Fatalf("want [item-b.versions], got %v", names)
+	}
+}

weed/filer/leveldb2/leveldb2_store.go

@@ -239,7 +239,7 @@ func (store *LevelDB2Store) ListDirectoryPrefixedEntries(ctx context.Context, di
 
 	directoryPrefix, partitionId := genDirectoryKeyPrefix(dirPath, prefix, store.dbCount)
 	lastFileStart := directoryPrefix
-	if startFileName != "" {
+	if startFileName > prefix {
 		lastFileStart, _ = genDirectoryKeyPrefix(dirPath, startFileName, store.dbCount)
 	}
 

weed/filer/leveldb2/leveldb2_store_test.go

@@ -151,6 +151,31 @@ func TestInitializeRejectsOversizedOption(t *testing.T) {
 	}
 }
 
+func TestListDirectoryPrefixedEntries_StartBelowPrefix(t *testing.T) {
+	store := &LevelDB2Store{}
+	store.initialize(t.TempDir(), 2)
+	defer store.Shutdown()
+
+	ctx := context.Background()
+	for _, name := range []string{"item-a", "item-a.versions", "item-b.versions"} {
+		if err := store.InsertEntry(ctx, &filer.Entry{FullPath: util.FullPath("/d/" + name)}); err != nil {
+			t.Fatalf("insert %s: %v", name, err)
+		}
+	}
+
+	var names []string
+	_, err := store.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), "item", true, 100, "item-b", func(entry *filer.Entry) (bool, error) {
+		names = append(names, entry.Name())
+		return true, nil
+	})
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	if len(names) != 1 || names[0] != "item-b.versions" {
+		t.Fatalf("want [item-b.versions], got %v", names)
+	}
+}
+
 func TestEmptyRoot(t *testing.T) {
 	testFiler := filer.NewFiler(pb.ServerDiscovery{}, nil, "", "", "", "", "", 255, nil)
 	dir := t.TempDir()

weed/filer/leveldb3/leveldb3_store.go

@@ -370,7 +370,7 @@ func (store *LevelDB3Store) ListDirectoryPrefixedEntries(ctx context.Context, di
 
 	directoryPrefix := genDirectoryKeyPrefix(shortPath, prefix)
 	lastFileStart := directoryPrefix
-	if startFileName != "" {
+	if startFileName > prefix {
 		lastFileStart = genDirectoryKeyPrefix(shortPath, startFileName)
 	}
 

weed/filer/leveldb3/leveldb3_store_test.go

@@ -182,3 +182,28 @@ func TestEmptyRoot(t *testing.T) {
 	}
 
 }
+
+func TestListDirectoryPrefixedEntries_StartBelowPrefix(t *testing.T) {
+	store := &LevelDB3Store{}
+	store.initialize(t.TempDir())
+	defer store.Shutdown()
+
+	ctx := context.Background()
+	for _, name := range []string{"item-a", "item-a.versions", "item-b.versions"} {
+		if err := store.InsertEntry(ctx, &filer.Entry{FullPath: util.FullPath("/d/" + name)}); err != nil {
+			t.Fatalf("insert %s: %v", name, err)
+		}
+	}
+
+	var names []string
+	_, err := store.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), "item", true, 100, "item-b", func(entry *filer.Entry) (bool, error) {
+		names = append(names, entry.Name())
+		return true, nil
+	})
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	if len(names) != 1 || names[0] != "item-b.versions" {
+		t.Fatalf("want [item-b.versions], got %v", names)
+	}
+}

weed/filer/redis2/universal_redis_store.go

@@ -15,6 +15,10 @@ import (
 
 const (
 	DIR_LIST_MARKER = "\x00"
+
+	// a listing whose callback stops early reads at most one batch more than
+	// it uses, and no single pipeline grows with the listing limit
+	listFetchBatchSize = 128
 )
 
 type UniversalRedis2Store struct {
@@ -96,7 +100,12 @@ func (store *UniversalRedis2Store) UpdateEntry(ctx context.Context, entry *filer
 
 func (store *UniversalRedis2Store) FindEntry(ctx context.Context, fullpath util.FullPath) (entry *filer.Entry, err error) {
 
-	data, err := store.Client.Get(ctx, store.getKey(string(fullpath))).Result()
+	return decodeEntry(fullpath, store.Client.Get(ctx, store.getKey(string(fullpath))))
+}
+
+func decodeEntry(fullpath util.FullPath, get *redis.StringCmd) (entry *filer.Entry, err error) {
+
+	data, err := get.Result()
 	if err == redis.Nil {
 		return nil, filer_pb.ErrNotFound
 	}
@@ -202,9 +211,13 @@ func (store *UniversalRedis2Store) ListDirectoryEntries(ctx context.Context, dir
 
 	// fetch entry meta
 	var entry *filer.Entry
-	for _, fileName := range members {
+	var gets []*redis.StringCmd
+	for i, fileName := range members {
+		if i%listFetchBatchSize == 0 {
+			gets = store.getEntries(ctx, dirPath, members[i:])
+		}
 		path := util.NewFullPath(string(dirPath), fileName)
-		entry, err = store.FindEntry(ctx, path)
+		entry, err = decodeEntry(path, gets[i%listFetchBatchSize])
 		lastFileName = fileName
 		if err != nil {
 			glog.V(0).InfofCtx(ctx, "list %s : %v", path, err)
@@ -235,6 +248,23 @@ func (store *UniversalRedis2Store) ListDirectoryEntries(ctx context.Context, dir
 	return lastFileName, err
 }
 
+// getEntries pipelines the GETs for up to listFetchBatchSize of members. Each
+// command keeps its own error, so a missing value still reads as redis.Nil and
+// fails only that member.
+func (store *UniversalRedis2Store) getEntries(ctx context.Context, dirPath util.FullPath, members []string) []*redis.StringCmd {
+	if len(members) > listFetchBatchSize {
+		members = members[:listFetchBatchSize]
+	}
+	gets := make([]*redis.StringCmd, len(members))
+	store.Client.Pipelined(ctx, func(pipe redis.Pipeliner) error {
+		for i, fileName := range members {
+			gets[i] = pipe.Get(ctx, store.getKey(string(util.NewFullPath(string(dirPath), fileName))))
+		}
+		return nil
+	})
+	return gets
+}
+
 func (store *UniversalRedis2Store) removeOrphanedDirectoryListMember(ctx context.Context, dirPath util.FullPath, fileName string) {
 	// a directory converted to super large after accumulating members still has a legacy index
 	if store.isSuperLargeDirectory(string(dirPath)) {

weed/filer/redis2/universal_redis_store_test.go

@@ -256,6 +256,95 @@ func TestListDirectoryEntriesDeletesLogicallyExpiredEntries(t *testing.T) {
 	}
 }
 
+func TestListDirectoryEntriesSkipsMissingAndExpiredMembersInOrder(t *testing.T) {
+	for _, keyPrefix := range []string{"", "sw:"} {
+		t.Run("keyPrefix="+keyPrefix, func(t *testing.T) {
+			store, dir := newTestStore(t, keyPrefix)
+
+			insertTestEntry(t, store, dir.Child("a"), 0)
+			insertTestEntry(t, store, dir.Child("b"), 0)
+			insertLogicallyExpiredTestEntry(t, store, dir.Child("c"))
+			insertTestEntry(t, store, dir.Child("d"), 0)
+
+			if err := store.Client.Del(context.Background(), store.getKey(string(dir.Child("b")))).Err(); err != nil {
+				t.Fatalf("drop value key: %v", err)
+			}
+
+			if names := listNames(t, store, dir); fmt.Sprint(names) != "[a d]" {
+				t.Fatalf("listed %v, want [a d]", names)
+			}
+
+			if members := indexMembers(t, store, dir); fmt.Sprint(members) != "[a d]" {
+				t.Fatalf("directory index holds %v, want [a d]", members)
+			}
+		})
+	}
+}
+
+func TestListDirectoryEntriesAcrossFetchBatches(t *testing.T) {
+	store, dir := newTestStore(t, "")
+
+	var want []string
+	for i := 0; i < listFetchBatchSize+2; i++ {
+		name := fmt.Sprintf("f%04d", i)
+		insertTestEntry(t, store, dir.Child(name), 0)
+		if i != listFetchBatchSize {
+			want = append(want, name)
+		}
+	}
+
+	// the first member of the second batch
+	missing := dir.Child(fmt.Sprintf("f%04d", listFetchBatchSize))
+	if err := store.Client.Del(context.Background(), store.getKey(string(missing))).Err(); err != nil {
+		t.Fatalf("drop value key: %v", err)
+	}
+
+	var names []string
+	if _, err := store.ListDirectoryEntries(context.Background(), dir, "", true, 1000, func(entry *filer.Entry) (bool, error) {
+		names = append(names, entry.Name())
+		return true, nil
+	}); err != nil {
+		t.Fatalf("ListDirectoryEntries %s: %v", dir, err)
+	}
+	if fmt.Sprint(names) != fmt.Sprint(want) {
+		t.Fatalf("listed %v, want %v", names, want)
+	}
+
+	if members := indexMembers(t, store, dir); fmt.Sprint(members) != fmt.Sprint(want) {
+		t.Fatalf("directory index holds %v, want %v", members, want)
+	}
+}
+
+func TestListDirectoryEntriesStopsAtUnreadableMember(t *testing.T) {
+	store, dir := newTestStore(t, "")
+	ctx := context.Background()
+
+	insertTestEntry(t, store, dir.Child("a"), 0)
+	insertTestEntry(t, store, dir.Child("b"), 0)
+	insertTestEntry(t, store, dir.Child("c"), 0)
+
+	// a value of the wrong type makes its GET fail with something other than redis.Nil
+	key := store.getKey(string(dir.Child("b")))
+	if err := store.Client.Del(ctx, key).Err(); err != nil {
+		t.Fatalf("drop value key: %v", err)
+	}
+	if err := store.Client.HSet(ctx, key, "f", "v").Err(); err != nil {
+		t.Fatalf("plant hash: %v", err)
+	}
+
+	var names []string
+	lastFileName, err := store.ListDirectoryEntries(ctx, dir, "", true, 100, func(entry *filer.Entry) (bool, error) {
+		names = append(names, entry.Name())
+		return true, nil
+	})
+	if err == nil {
+		t.Fatal("ListDirectoryEntries returned no error for an unreadable member")
+	}
+	if fmt.Sprint(names) != "[a]" || lastFileName != "b" {
+		t.Fatalf("listed %v up to %q, want [a] up to \"b\"", names, lastFileName)
+	}
+}
+
 func TestDeleteExpiredEntryKeepsRecreatedValue(t *testing.T) {
 	store, dir := newTestStore(t, "")
 

weed/filer/rocksdb/rocksdb_store.go

@@ -243,7 +243,7 @@ func (store *RocksDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dir
 
 	directoryPrefix := genDirectoryKeyPrefix(dirPath, prefix)
 	lastFileStart := directoryPrefix
-	if startFileName != "" {
+	if startFileName > prefix {
 		lastFileStart = genDirectoryKeyPrefix(dirPath, startFileName)
 	}
 

weed/filer/tikv/tikv_store.go

@@ -246,7 +246,7 @@ func (store *TikvStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPat
 	lastFileName := ""
 	directoryPrefix := store.genDirectoryKeyPrefix(dirPath, prefix)
 	lastFileStart := directoryPrefix
-	if startFileName != "" {
+	if startFileName > prefix {
 		lastFileStart = store.genDirectoryKeyPrefix(dirPath, startFileName)
 	}
 

weed/s3api/s3api_object_handlers_postpolicy.go

@@ -139,7 +139,7 @@ func (s3a *S3ApiServer) PostPolicyBucketHandler(w http.ResponseWriter, r *http.R
 	// Forward validated POST form fields to the underlying PUT as headers.
 	applyPostPolicyFormHeaders(r, formValues)
 
-	// Authorize the object like the PUT path; the coarse Write check above does not consult the bucket policy.
+	// Authorize the object like the PUT path.
 	if errCode := s3a.iam.AuthorizeObjectWrite(r, identity, bucket, object); errCode != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, r, refineDenialForBucketName(r, identity, errCode))
 		return
@@ -382,25 +382,15 @@ func getRedirectPostRawQuery(bucket, key, etag string) string {
 }
 
 // Check to see if Policy is signed correctly, returning the signing identity.
-// The verified form's access key is recorded in ctx's identity holder. The
-// identity's Write permission on the bucket is checked only once the signature
-// verifies, as on the header-signed path, so a form that fails verification
-// learns nothing about the key's permissions. A verified identity that may not
-// write is returned with ErrAccessDenied.
+// The verified form's access key is recorded in ctx's identity holder.
+// Write authorization happens later in the handler via AuthorizeObjectWrite,
+// which evaluates the identity's attached IAM policies, session policies, and
+// the bucket policy against the real object key; gating here on the legacy
+// Actions projection would deny identities those mechanisms allow.
 func (iam *IdentityAccessManagement) doesPolicySignatureMatch(ctx context.Context, formValues http.Header) (*Identity, s3err.ErrorCode) {
-	var identity *Identity
-	var errCode s3err.ErrorCode
 	// For SignV2 - Signature field will be valid
 	if _, ok := formValues["Signature"]; ok {
-		identity, errCode = iam.doesPolicySignatureV2Match(ctx, formValues)
-	} else {
-		identity, errCode = iam.doesPolicySignatureV4Match(ctx, formValues)
+		return iam.doesPolicySignatureV2Match(ctx, formValues)
 	}
-	if errCode != s3err.ErrNone {
-		return nil, errCode
-	}
-	if !identity.CanDo(s3_constants.ACTION_WRITE, formValues.Get("bucket"), "") {
-		return identity, s3err.ErrAccessDenied
-	}
-	return identity, s3err.ErrNone
+	return iam.doesPolicySignatureV4Match(ctx, formValues)
 }

weed/s3api/s3api_object_handlers_postpolicy_authz_test.go

@@ -32,6 +32,11 @@ const (
 	postAclAccessKey = "AKIAACLTESTTEST"
 	postAclSecretKey = "secret-key-for-acl-tests"
 	postAclPrincipal = "arn:aws:iam::000000000000:user/aclwriter"
+
+	postPolicyAccessKey = "AKIAPOLICYONLY0000"
+	postPolicySecretKey = "secret-key-for-policy-only"
+	postReadAccessKey   = "AKIAPOLICYREAD0000"
+	postReadSecretKey   = "secret-key-for-policy-read"
 )
 
 // newPostAuthzServer builds an S3ApiServer whose non-admin "tester" identity
@@ -274,3 +279,118 @@ func TestAuthorizeObjectAclWrite(t *testing.T) {
 		s3a.iam.AuthorizeObjectAclWrite(req, admin, postAuthzBucket, "acl-denied/secret.txt"),
 		"an admin is not confined by the bucket policy")
 }
+
+// newPostPolicyOnlyServer builds an S3ApiServer with identities that hold no
+// legacy Write action: "poster" carries an attached IAM policy allowing
+// s3:PutObject on the test bucket, and "reader" one allowing only s3:GetObject.
+func newPostPolicyOnlyServer(t *testing.T) *S3ApiServer {
+	t.Helper()
+
+	iam := &IdentityAccessManagement{
+		hashes:       make(map[string]*sync.Pool),
+		hashCounters: make(map[string]*int32),
+	}
+	err := iam.loadS3ApiConfiguration(&iam_pb.S3ApiConfiguration{
+		Identities: []*iam_pb.Identity{{
+			Name:        "poster",
+			Account:     &iam_pb.Account{Id: "000000000000", DisplayName: "poster"},
+			Credentials: []*iam_pb.Credential{{AccessKey: postPolicyAccessKey, SecretKey: postPolicySecretKey}},
+			Actions:     []string{"List:" + postAuthzBucket},
+			PolicyNames: []string{"post-write"},
+		}, {
+			Name:        "reader",
+			Account:     &iam_pb.Account{Id: "000000000000", DisplayName: "reader"},
+			Credentials: []*iam_pb.Credential{{AccessKey: postReadAccessKey, SecretKey: postReadSecretKey}},
+			Actions:     []string{"Read:" + postAuthzBucket, "List:" + postAuthzBucket},
+			PolicyNames: []string{"post-read"},
+		}},
+		Policies: []*iam_pb.Policy{{
+			Name: "post-write",
+			Content: fmt.Sprintf(`{
+				"Version": "2012-10-17",
+				"Statement": [{
+					"Effect": "Allow",
+					"Action": ["s3:PutObject", "s3:GetObject", "s3:DeleteObject"],
+					"Resource": ["arn:aws:s3:::%s/*"]
+				}, {
+					"Effect": "Allow",
+					"Action": ["s3:ListBucket", "s3:GetBucketLocation"],
+					"Resource": ["arn:aws:s3:::%s"]
+				}]
+			}`, postAuthzBucket, postAuthzBucket),
+		}, {
+			Name: "post-read",
+			Content: fmt.Sprintf(`{
+				"Version": "2012-10-17",
+				"Statement": [{
+					"Effect": "Allow",
+					"Action": ["s3:GetObject"],
+					"Resource": ["arn:aws:s3:::%s/*"]
+				}]
+			}`, postAuthzBucket),
+		}},
+	})
+	require.NoError(t, err)
+
+	s3a := &S3ApiServer{
+		option: &S3ApiServerOption{BucketsPath: "/buckets"},
+		iam:    iam,
+	}
+	s3a.bucketRegistry = NewBucketRegistry(s3a)
+	s3a.bucketRegistry.setMetadataCache(&BucketMetaData{Name: postAuthzBucket, IsTableBucket: false})
+	return s3a
+}
+
+// postFormValues canonicalizes the signed request's multipart fields the way
+// extractPostPolicyFormValues does, for driving doesPolicySignatureMatch.
+func postFormValues(t *testing.T, r *http.Request) http.Header {
+	t.Helper()
+	require.NoError(t, r.ParseMultipartForm(1<<20))
+	formValues := make(http.Header)
+	for k, v := range r.MultipartForm.Value {
+		formValues[http.CanonicalHeaderKey(k)] = v
+	}
+	return formValues
+}
+
+// TestPostPolicySignatureAllowsAttachedPolicyIdentity signs a POST Object form
+// with a key whose holder has no legacy Write action - only an attached IAM
+// policy granting s3:PutObject - and asserts the signature check accepts it
+// and the shared object-write authorization allows the key. Gating the
+// signature check on the Actions projection used to fail this with
+// AccessDenied before authorization ever ran.
+func TestPostPolicySignatureAllowsAttachedPolicyIdentity(t *testing.T) {
+	s3a := newPostPolicyOnlyServer(t)
+
+	req := newSignedPostRequestFor(t, "uploads/pic.png", postPolicyAccessKey, postPolicySecretKey, nil)
+	identity, errCode := s3a.iam.doesPolicySignatureMatch(req.Context(), postFormValues(t, req))
+	require.Equal(t, s3err.ErrNone, errCode,
+		"a verified POST signature must not require the legacy Write action")
+	require.NotNil(t, identity)
+
+	assert.Equal(t, s3err.ErrNone,
+		s3a.iam.AuthorizeObjectWrite(req, identity, postAuthzBucket, "uploads/pic.png"),
+		"an attached IAM policy granting s3:PutObject must authorize the key like the PUT path")
+}
+
+// TestPostPolicyAttachedPolicyDeny covers the deny leg: a verified identity
+// whose attached policy does not grant s3:PutObject is still refused by
+// AuthorizeObjectWrite and by the full handler.
+func TestPostPolicyAttachedPolicyDeny(t *testing.T) {
+	s3a := newPostPolicyOnlyServer(t)
+
+	req := newSignedPostRequestFor(t, "uploads/pic.png", postReadAccessKey, postReadSecretKey, nil)
+	identity, errCode := s3a.iam.doesPolicySignatureMatch(req.Context(), postFormValues(t, req))
+	require.Equal(t, s3err.ErrNone, errCode)
+	require.NotNil(t, identity)
+
+	assert.Equal(t, s3err.ErrAccessDenied,
+		s3a.iam.AuthorizeObjectWrite(req, identity, postAuthzBucket, "uploads/pic.png"),
+		"an attached policy without s3:PutObject must still deny the upload")
+
+	rec := httptest.NewRecorder()
+	s3a.PostPolicyBucketHandler(rec, newSignedPostRequestFor(t, "uploads/pic.png", postReadAccessKey, postReadSecretKey, nil))
+	assert.Equal(t, http.StatusForbidden, rec.Code,
+		"POST from a read-only identity must stay AccessDenied, body: %s", rec.Body.String())
+	assert.Contains(t, rec.Body.String(), "AccessDenied")
+}

weed/shell/command_volume_configure_replication.go

@@ -6,6 +6,7 @@ import (
 	"flag"
 	"fmt"
 	"io"
+	"sync"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 
@@ -33,6 +34,9 @@ func (c *commandVolumeConfigureReplication) Help() string {
 
 	This command changes a volume replication value. It should be followed by "volume.fix.replication".
 
+	Up to -maxParallelization volume servers are configured at a time, one volume at a time on each,
+	and replicas of the same volume on different servers take turns.
+
 `
 }
 
@@ -46,6 +50,7 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman
 	volumeIdInt := configureReplicationCommand.Int("volumeId", 0, "the volume id")
 	replicationString := configureReplicationCommand.String("replication", "", "the intended replication value")
 	collectionPattern := configureReplicationCommand.String("collectionPattern", "", "comma-separated collection names, with '*' and '?' wildcards; empty matches all")
+	maxParallelization := configureReplicationCommand.Int("maxParallelization", DefaultMaxParallelization, "run up to X tasks in parallel, whenever possible")
 	if err = configureReplicationCommand.Parse(args); err != nil {
 		return nil
 	}
@@ -75,6 +80,19 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman
 		return err
 	}
 
+	return configureReplicationOnVolumeServers(commandEnv, topologyInfo, volumeFilter, replicaPlacement.String(), *maxParallelization)
+}
+
+func configureReplicationOnVolumeServers(commandEnv *CommandEnv, topologyInfo *master_pb.TopologyInfo, volumeFilter func(*master_pb.VolumeInformationMessage) bool, replication string, maxParallelization int) error {
+	type serverVolumes struct {
+		addr      pb.ServerAddress
+		volumeIds []uint32
+	}
+	var servers []serverVolumes
+	// VolumeConfigure unmounts and remounts the volume, so replicas take turns
+	// to keep at least one of them serving.
+	volumeLocks := make(map[uint32]*sync.Mutex)
+
 	// find all data nodes with volumes that needs replication change
 	eachDataNode(topologyInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
 		var targetVolumeIds []uint32
@@ -82,33 +100,45 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman
 			for _, v := range diskInfo.VolumeInfos {
 				if volumeFilter(v) {
 					targetVolumeIds = append(targetVolumeIds, v.Id)
+					if volumeLocks[v.Id] == nil {
+						volumeLocks[v.Id] = &sync.Mutex{}
+					}
 				}
 			}
 		}
-		if len(targetVolumeIds) == 0 {
-			return
-		}
-		err = operation.WithVolumeServerClient(false, pb.NewServerAddressFromDataNode(dn), commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
-			for _, targetVolumeId := range targetVolumeIds {
-				resp, configureErr := volumeServerClient.VolumeConfigure(context.Background(), &volume_server_pb.VolumeConfigureRequest{
-					VolumeId:    targetVolumeId,
-					Replication: replicaPlacement.String(),
-				})
-				if configureErr != nil {
-					return configureErr
-				}
-				if resp.Error != "" {
-					return errors.New(resp.Error)
-				}
-			}
-			return nil
-		})
-		if err != nil {
-			return
+		if len(targetVolumeIds) > 0 {
+			servers = append(servers, serverVolumes{pb.NewServerAddressFromDataNode(dn), targetVolumeIds})
 		}
 	})
 
-	return err
+	ewg := NewErrorWaitGroup(maxParallelization)
+	for _, server := range servers {
+		ewg.Add(func() error {
+			err := operation.WithVolumeServerClient(false, server.addr, commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
+				for _, targetVolumeId := range server.volumeIds {
+					volumeLocks[targetVolumeId].Lock()
+					resp, configureErr := volumeServerClient.VolumeConfigure(context.Background(), &volume_server_pb.VolumeConfigureRequest{
+						VolumeId:    targetVolumeId,
+						Replication: replication,
+					})
+					volumeLocks[targetVolumeId].Unlock()
+					if configureErr != nil {
+						return configureErr
+					}
+					if resp.Error != "" {
+						return errors.New(resp.Error)
+					}
+				}
+				return nil
+			})
+			if err != nil {
+				return fmt.Errorf("volume server %s: %w", server.addr, err)
+			}
+			return nil
+		})
+	}
+
+	return ewg.Wait()
 }
 
 func getVolumeFilter(replicaPlacement *super_block.ReplicaPlacement, volumeId uint32, collectionPattern string) (func(message *master_pb.VolumeInformationMessage) bool, error) {

weed/shell/command_volume_configure_replication_test.go

@@ -0,0 +1,226 @@
+package shell
+
+import (
+	"context"
+	"fmt"
+	"maps"
+	"net"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+type configureReplicationTestVolumeServer struct {
+	volume_server_pb.UnimplementedVolumeServerServer
+	failure    string
+	mu         sync.Mutex
+	configured map[uint32]string
+}
+
+func (s *configureReplicationTestVolumeServer) VolumeConfigure(_ context.Context, req *volume_server_pb.VolumeConfigureRequest) (*volume_server_pb.VolumeConfigureResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.configured == nil {
+		s.configured = make(map[uint32]string)
+	}
+	s.configured[req.VolumeId] = req.Replication
+	return &volume_server_pb.VolumeConfigureResponse{Error: s.failure}, nil
+}
+
+// configureReplicationOverlapServer can back several volume servers at once and
+// records whether two calls for the same volume were ever in flight together.
+type configureReplicationOverlapServer struct {
+	volume_server_pb.UnimplementedVolumeServerServer
+	mu         sync.Mutex
+	inFlight   map[uint32]int
+	overlapped bool
+}
+
+func (s *configureReplicationOverlapServer) VolumeConfigure(_ context.Context, req *volume_server_pb.VolumeConfigureRequest) (*volume_server_pb.VolumeConfigureResponse, error) {
+	s.mu.Lock()
+	s.inFlight[req.VolumeId]++
+	if s.inFlight[req.VolumeId] > 1 {
+		s.overlapped = true
+	}
+	s.mu.Unlock()
+
+	time.Sleep(20 * time.Millisecond)
+
+	s.mu.Lock()
+	s.inFlight[req.VolumeId]--
+	s.mu.Unlock()
+	return &volume_server_pb.VolumeConfigureResponse{}, nil
+}
+
+func startConfigureReplicationTestVolumeServer(t *testing.T, server volume_server_pb.VolumeServerServer, volumeIds ...uint32) *master_pb.DataNodeInfo {
+	t.Helper()
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	srv := grpc.NewServer()
+	volume_server_pb.RegisterVolumeServerServer(srv, server)
+	go srv.Serve(lis)
+	t.Cleanup(srv.Stop)
+
+	port := lis.Addr().(*net.TCPAddr).Port
+	disk := &master_pb.DiskInfo{}
+	for _, vid := range volumeIds {
+		disk.VolumeInfos = append(disk.VolumeInfos, &master_pb.VolumeInformationMessage{Id: vid})
+	}
+	return &master_pb.DataNodeInfo{
+		Id:        fmt.Sprintf("127.0.0.1:%d", port),
+		GrpcPort:  uint32(port),
+		DiskInfos: map[string]*master_pb.DiskInfo{"": disk},
+	}
+}
+
+func configureReplicationTestEnv(dataNodes ...*master_pb.DataNodeInfo) (*CommandEnv, *master_pb.TopologyInfo) {
+	commandEnv := &CommandEnv{option: &ShellOptions{
+		GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
+	}}
+	topologyInfo := &master_pb.TopologyInfo{DataCenterInfos: []*master_pb.DataCenterInfo{{
+		RackInfos: []*master_pb.RackInfo{{DataNodeInfos: dataNodes}},
+	}}}
+	return commandEnv, topologyInfo
+}
+
+func allVolumes(*master_pb.VolumeInformationMessage) bool { return true }
+
+func TestConfigureReplicationOnVolumeServersReportsEveryFailingServer(t *testing.T) {
+	for _, maxParallelization := range []int{1, DefaultMaxParallelization} {
+		t.Run(fmt.Sprintf("maxParallelization=%d", maxParallelization), func(t *testing.T) {
+			first := &configureReplicationTestVolumeServer{failure: "first server failed"}
+			healthy := &configureReplicationTestVolumeServer{}
+			last := &configureReplicationTestVolumeServer{failure: "last server failed"}
+			commandEnv, topologyInfo := configureReplicationTestEnv(
+				startConfigureReplicationTestVolumeServer(t, first, 1),
+				startConfigureReplicationTestVolumeServer(t, healthy, 2, 3),
+				startConfigureReplicationTestVolumeServer(t, last, 4),
+			)
+
+			err := configureReplicationOnVolumeServers(commandEnv, topologyInfo, allVolumes, "001", maxParallelization)
+
+			if err == nil {
+				t.Fatal("expected an error")
+			}
+			for _, failure := range []string{first.failure, last.failure} {
+				if !strings.Contains(err.Error(), failure) {
+					t.Errorf("error %q does not mention %q", err, failure)
+				}
+			}
+			healthy.mu.Lock()
+			defer healthy.mu.Unlock()
+			if expected := map[uint32]string{2: "001", 3: "001"}; !maps.Equal(healthy.configured, expected) {
+				t.Errorf("healthy server configured %v, expected %v", healthy.configured, expected)
+			}
+		})
+	}
+}
+
+func TestConfigureReplicationOnVolumeServersConfiguresReplicasOneAtATime(t *testing.T) {
+	server := &configureReplicationOverlapServer{inFlight: make(map[uint32]int)}
+	commandEnv, topologyInfo := configureReplicationTestEnv(
+		startConfigureReplicationTestVolumeServer(t, server, 1, 2),
+		startConfigureReplicationTestVolumeServer(t, server, 1, 2),
+		startConfigureReplicationTestVolumeServer(t, server, 1, 2),
+	)
+
+	if err := configureReplicationOnVolumeServers(commandEnv, topologyInfo, allVolumes, "002", DefaultMaxParallelization); err != nil {
+		t.Fatal(err)
+	}
+
+	server.mu.Lock()
+	defer server.mu.Unlock()
+	if server.overlapped {
+		t.Error("replicas of one volume were configured at the same time")
+	}
+}
+
+// configureReplicationGateServer blocks inside VolumeConfigure until the test
+// releases it, so the test can observe how many servers run at once.
+type configureReplicationGateServer struct {
+	volume_server_pb.UnimplementedVolumeServerServer
+	gate *configureReplicationGate
+}
+
+type configureReplicationGate struct {
+	entered   chan struct{}
+	release   chan struct{}
+	mu        sync.Mutex
+	active    int
+	maxActive int
+	calls     int
+}
+
+func (s *configureReplicationGateServer) VolumeConfigure(_ context.Context, _ *volume_server_pb.VolumeConfigureRequest) (*volume_server_pb.VolumeConfigureResponse, error) {
+	s.gate.mu.Lock()
+	s.gate.calls++
+	s.gate.active++
+	if s.gate.active > s.gate.maxActive {
+		s.gate.maxActive = s.gate.active
+	}
+	s.gate.mu.Unlock()
+
+	s.gate.entered <- struct{}{}
+	<-s.gate.release
+
+	s.gate.mu.Lock()
+	s.gate.active--
+	s.gate.mu.Unlock()
+	return &volume_server_pb.VolumeConfigureResponse{}, nil
+}
+
+func TestConfigureReplicationOnVolumeServersRunsServersInParallelUpToTheLimit(t *testing.T) {
+	const servers = 4
+	const maxParallelization = 2
+	gate := &configureReplicationGate{
+		entered: make(chan struct{}, servers),
+		release: make(chan struct{}),
+	}
+	var dataNodes []*master_pb.DataNodeInfo
+	for i := 0; i < servers; i++ {
+		dataNodes = append(dataNodes, startConfigureReplicationTestVolumeServer(t,
+			&configureReplicationGateServer{gate: gate}, uint32(i+1)))
+	}
+	commandEnv, topologyInfo := configureReplicationTestEnv(dataNodes...)
+
+	done := make(chan error, 1)
+	go func() {
+		done <- configureReplicationOnVolumeServers(commandEnv, topologyInfo, allVolumes, "001", maxParallelization)
+	}()
+
+	for i := 0; i < maxParallelization; i++ {
+		select {
+		case <-gate.entered:
+		case <-time.After(10 * time.Second):
+			t.Fatalf("only %d servers started; serial execution would hold the second call", i)
+		}
+	}
+
+	select {
+	case <-gate.entered:
+		t.Fatalf("more than %d servers ran while calls were held open", maxParallelization)
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	close(gate.release)
+	if err := <-done; err != nil {
+		t.Fatal(err)
+	}
+
+	gate.mu.Lock()
+	defer gate.mu.Unlock()
+	if gate.maxActive != maxParallelization {
+		t.Errorf("at most %d servers ran together, expected %d", gate.maxActive, maxParallelization)
+	}
+	if gate.calls != servers {
+		t.Errorf("%d servers ran, expected %d", gate.calls, servers)
+	}
+}

weed/topology/store_replicate.go

@@ -9,6 +9,7 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -46,8 +47,8 @@ func ReplicatedWrite(ctx context.Context, masterFn operation.GetMasterFn, grpcDi
 	}
 
 	// with no local volume and no replica to forward to, the write would be acknowledged without being stored
-	hasLocalVolume := s.HasVolume(volumeId)
-	if !hasLocalVolume && len(remoteLocations) == 0 {
+	vol := s.GetVolume(volumeId)
+	if vol == nil && len(remoteLocations) == 0 {
 		err = fmt.Errorf("volume %d not found on %s:%d", volumeId, s.Ip, s.Port)
 		glog.V(0).Infoln(err)
 		return
@@ -68,109 +69,141 @@ func ReplicatedWrite(ctx context.Context, masterFn operation.GetMasterFn, grpcDi
 		}(time.Now())
 	}
 
-	if hasLocalVolume {
-		start := time.Now()
-
-		inFlightGauge := stats.VolumeServerInFlightRequestsGauge.WithLabelValues(stats.WriteToLocalDisk)
-		inFlightGauge.Inc()
-		defer inFlightGauge.Dec()
-
-		isUnchanged, err = s.WriteVolumeNeedle(volumeId, n, true, fsync)
-		stats.VolumeServerRequestHistogram.WithLabelValues(stats.WriteToLocalDisk).Observe(time.Since(start).Seconds())
-		if err != nil {
-			stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorWriteToLocalDisk).Inc()
-			err = fmt.Errorf("failed to write to local disk: %w", err)
-			glog.V(0).Infoln(err)
-			return
+	// Snapshot the fields the fan-out reads before the concurrent local write
+	// mutates the needle (it stamps the volume TTL and flags), so the two cannot
+	// race; the fan-out then touches only these locals plus read-only n.Data.
+	replicaTtl := n.Ttl
+	if vol != nil && replicaTtl == needle.EMPTY_TTL && vol.Ttl != needle.EMPTY_TTL {
+		replicaTtl = vol.Ttl
+	}
+	replicaLastModified := n.LastModified
+	replicaIsChunked := n.IsChunkedManifest()
+	replicaIsCompressed := n.IsCompressed()
+	replicaName := string(n.Name)
+	replicaMime := string(n.Mime)
+	replicaData := n.Data
+	replicaPairMap := make(map[string]string)
+	if replicaCount > 0 && n.HasPairs() {
+		tmpMap := make(map[string]string)
+		if pe := json.Unmarshal(n.Pairs, &tmpMap); pe != nil {
+			stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorUnmarshalPairs).Inc()
+			glog.V(0).Infoln("Unmarshal pairs error:", pe)
+		}
+		for k, v := range tmpMap {
+			replicaPairMap[needle.PairNamePrefix+k] = v
 		}
 	}
 
 	// Observe replication targets histogram for all operations (including zero)
 	stats.VolumeServerReplicationTargets.Observe(float64(replicaCount))
 
-	if replicaCount > 0 { //send to other replica locations
-		start := time.Now()
+	// Run the local disk write and the replica fan-out concurrently against the
+	// buffered needle (latency max(local, replicas), not the sum); ack after both.
+	var wg sync.WaitGroup
+	var localErr, remoteErr error
+	var localIsUnchanged bool
 
-		inFlightGauge := stats.VolumeServerInFlightRequestsGauge.WithLabelValues(stats.WriteToReplicas)
-		inFlightGauge.Inc()
-		defer inFlightGauge.Dec()
+	if vol != nil {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			start := time.Now()
+			inFlightGauge := stats.VolumeServerInFlightRequestsGauge.WithLabelValues(stats.WriteToLocalDisk)
+			inFlightGauge.Inc()
+			defer inFlightGauge.Dec()
 
-		err = DistributedOperation(ctx, remoteLocations, func(ctx context.Context, location operation.Location) error {
-			u := url.URL{
-				Scheme: "http",
-				Host:   location.Url,
-				Path:   r.URL.Path,
+			localIsUnchanged, localErr = s.WriteVolumeNeedle(volumeId, n, true, fsync)
+			stats.VolumeServerRequestHistogram.WithLabelValues(stats.WriteToLocalDisk).Observe(time.Since(start).Seconds())
+			if localErr != nil {
+				stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorWriteToLocalDisk).Inc()
+				localErr = fmt.Errorf("failed to write to local disk: %w", localErr)
+				glog.V(0).Infoln(localErr)
 			}
-			q := url.Values{
-				"type": {"replicate"},
-				"ttl":  {n.Ttl.String()},
-			}
-			if fsync {
-				q.Set("fsync", "true")
-			}
-			if n.LastModified > 0 {
-				q.Set("ts", strconv.FormatUint(n.LastModified, 10))
-			}
-			if n.IsChunkedManifest() {
-				q.Set("cm", "true")
-			}
-			u.RawQuery = q.Encode()
-
-			pairMap := make(map[string]string)
-			if n.HasPairs() {
-				tmpMap := make(map[string]string)
-				err := json.Unmarshal(n.Pairs, &tmpMap)
-				if err != nil {
-					stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorUnmarshalPairs).Inc()
-					glog.V(0).Infoln("Unmarshal pairs error:", err)
-				}
-				for k, v := range tmpMap {
-					pairMap[needle.PairNamePrefix+k] = v
-				}
-			}
-			bytesBuffer := buffer_pool.SyncPoolGetBuffer()
-			defer buffer_pool.SyncPoolPutBuffer(bytesBuffer)
-
-			// volume server do not know about encryption
-			// TODO optimize here to compress data only once
-			uploadOption := &operation.UploadOption{
-				UploadUrl:         u.String(),
-				Filename:          string(n.Name),
-				Cipher:            false,
-				IsInputCompressed: n.IsCompressed(),
-				IsReplication:     true,
-				MimeType:          string(n.Mime),
-				PairMap:           pairMap,
-				Jwt:               jwt,
-				Md5:               contentMd5,
-				BytesBuffer:       bytesBuffer,
-				MaxAttempts:       1, // fail fast on a dead replica; the client write retries
-			}
-
-			uploader, err := operation.NewUploader()
-			if err != nil {
-				glog.Errorf("replication-UploadData, err:%v, url:%s", err, u.String())
-				return err
-			}
-			_, err = uploader.UploadData(ctx, n.Data, uploadOption)
-			if err != nil {
-				glog.Errorf("replication-UploadData, err:%v, url:%s", err, u.String())
-			}
-			return err
-		})
-		stats.VolumeServerRequestHistogram.WithLabelValues(stats.WriteToReplicas).Observe(time.Since(start).Seconds())
-		if err != nil {
-			stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorWriteToReplicas).Inc()
-			stats.VolumeServerReplicationCounter.WithLabelValues(stats.ReplicationOpWrite, stats.ReplicationFailure).Inc()
-			reason := classifyReplicationError(err)
-			stats.VolumeServerReplicationFailures.WithLabelValues(stats.ReplicationOpWrite, reason).Inc()
-			err = fmt.Errorf("failed to write to replicas for volume %d: %v", volumeId, err)
-			glog.V(0).Infoln(err)
-			return false, err
-		}
-		stats.VolumeServerReplicationCounter.WithLabelValues(stats.ReplicationOpWrite, stats.ReplicationSuccess).Inc()
+		}()
 	}
-	return
+
+	if replicaCount > 0 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			start := time.Now()
+			inFlightGauge := stats.VolumeServerInFlightRequestsGauge.WithLabelValues(stats.WriteToReplicas)
+			inFlightGauge.Inc()
+			defer inFlightGauge.Dec()
+
+			remoteErr = DistributedOperation(ctx, remoteLocations, func(ctx context.Context, location operation.Location) error {
+				u := url.URL{
+					Scheme: "http",
+					Host:   location.Url,
+					Path:   r.URL.Path,
+				}
+				q := url.Values{
+					"type": {"replicate"},
+					"ttl":  {replicaTtl.String()},
+				}
+				if fsync {
+					q.Set("fsync", "true")
+				}
+				if replicaLastModified > 0 {
+					q.Set("ts", strconv.FormatUint(replicaLastModified, 10))
+				}
+				if replicaIsChunked {
+					q.Set("cm", "true")
+				}
+				u.RawQuery = q.Encode()
+
+				bytesBuffer := buffer_pool.SyncPoolGetBuffer()
+				defer buffer_pool.SyncPoolPutBuffer(bytesBuffer)
+
+				// volume server do not know about encryption
+				uploadOption := &operation.UploadOption{
+					UploadUrl:         u.String(),
+					Filename:          replicaName,
+					Cipher:            false,
+					IsInputCompressed: replicaIsCompressed,
+					IsReplication:     true,
+					MimeType:          replicaMime,
+					PairMap:           replicaPairMap,
+					Jwt:               jwt,
+					Md5:               contentMd5,
+					BytesBuffer:       bytesBuffer,
+					MaxAttempts:       1, // fail fast on a dead replica; the client write retries
+				}
+
+				uploader, uploaderErr := operation.NewUploader()
+				if uploaderErr != nil {
+					glog.Errorf("replication-UploadData, err:%v, url:%s", uploaderErr, u.String())
+					return uploaderErr
+				}
+				_, uploadErr := uploader.UploadData(ctx, replicaData, uploadOption)
+				if uploadErr != nil {
+					glog.Errorf("replication-UploadData, err:%v, url:%s", uploadErr, u.String())
+				}
+				return uploadErr
+			})
+			stats.VolumeServerRequestHistogram.WithLabelValues(stats.WriteToReplicas).Observe(time.Since(start).Seconds())
+			if remoteErr != nil {
+				stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorWriteToReplicas).Inc()
+				stats.VolumeServerReplicationCounter.WithLabelValues(stats.ReplicationOpWrite, stats.ReplicationFailure).Inc()
+				reason := classifyReplicationError(remoteErr)
+				stats.VolumeServerReplicationFailures.WithLabelValues(stats.ReplicationOpWrite, reason).Inc()
+				remoteErr = fmt.Errorf("failed to write to replicas for volume %d: %v", volumeId, remoteErr)
+				glog.V(0).Infoln(remoteErr)
+			} else {
+				stats.VolumeServerReplicationCounter.WithLabelValues(stats.ReplicationOpWrite, stats.ReplicationSuccess).Inc()
+			}
+		}()
+	}
+
+	wg.Wait()
+
+	if localErr != nil {
+		return false, localErr
+	}
+	if remoteErr != nil {
+		return false, remoteErr
+	}
+	return localIsUnchanged, nil
 }
 
 // ReplicatedDelete deletes a needle from the local volume and sends delete

weed/topology/store_replicate_test.go

@@ -16,8 +16,11 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/operation"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/storage"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 )
@@ -230,6 +233,176 @@ func TestReplicatedWriteForwardsFsyncToReplicas(t *testing.T) {
 	}
 }
 
+// newReplicateTestStore builds a single-disk store in dir, draining the notify
+// channels so volume mounts never block on a heartbeat reader.
+func newReplicateTestStore(t *testing.T, dir string) *storage.Store {
+	t.Helper()
+	store := storage.NewStore(nil, "127.0.0.1", 8080, 18080, "http://127.0.0.1:8080", "store-id",
+		[]string{dir}, []int32{100}, []util.MinFreeSpace{{}}, "",
+		storage.NeedleMapInMemory, []types.DiskType{types.HardDriveType}, nil, 3,
+		stats.DefaultDiskIOProbeConfig())
+	done := make(chan struct{})
+	go func() {
+		for {
+			select {
+			case <-store.NewVolumesChan:
+			case <-store.DeletedVolumesChan:
+			case <-store.NewEcShardsChan:
+			case <-store.DeletedEcShardsChan:
+			case <-store.StateUpdateChan:
+			case <-done:
+				return
+			}
+		}
+	}()
+	t.Cleanup(func() {
+		close(done)
+		store.Close()
+	})
+	return store
+}
+
+// TestReplicatedWriteOverlapsLocalAndReplicaWrites runs the write path with a
+// real local volume and a replica that stalls mid-request: the fan-out starts
+// while the local write is still free to run, the local write lands before the
+// replica unblocks, and the acknowledged write waits for both legs.
+func TestReplicatedWriteOverlapsLocalAndReplicaWrites(t *testing.T) {
+	replicaStarted := make(chan url.Values, 1)
+	releaseReplica := make(chan struct{})
+	var releaseOnce sync.Once
+	release := func() { releaseOnce.Do(func() { close(releaseReplica) }) }
+	replica := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		replicaStarted <- r.URL.Query()
+		<-releaseReplica
+		w.WriteHeader(http.StatusCreated)
+		if _, err := w.Write([]byte(`{"size":21}`)); err != nil {
+			t.Errorf("replica write response: %v", err)
+		}
+	}))
+	defer replica.Close()
+	defer release()
+	replicaHost := strings.TrimPrefix(replica.URL, "http://")
+
+	masterFn, dialOption := startMockMasterServer(t, &mockMasterServer{
+		locations: []*master_pb.Location{{Url: "127.0.0.1:8080"}, {Url: replicaHost}},
+	})
+
+	store := newReplicateTestStore(t, t.TempDir())
+	volumeId := needle.VolumeId(7)
+	operation.InvalidateVolumeIdLocationCache(volumeId.String())
+	if err := store.AddVolume(volumeId, "", storage.NeedleMapInMemory, "001", "15d", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil {
+		t.Fatal(err)
+	}
+
+	data := []byte("overlapped write data")
+	n := &needle.Needle{
+		Id:       1,
+		Data:     data,
+		Checksum: needle.NewCRC(data),
+		Ttl:      needle.EMPTY_TTL,
+	}
+	r := httptest.NewRequest(http.MethodPost, "http://127.0.0.1:8080/7,01637037d6", nil)
+	if err := r.ParseForm(); err != nil {
+		t.Fatal(err)
+	}
+
+	done := make(chan error, 1)
+	go func() {
+		_, err := ReplicatedWrite(context.Background(), masterFn, dialOption, store, volumeId, n, r, "")
+		done <- err
+	}()
+
+	var q url.Values
+	select {
+	case q = <-replicaStarted:
+	case <-time.After(5 * time.Second):
+		t.Fatal("replica never received the fan-out request")
+	}
+	if got := q.Get("type"); got != "replicate" {
+		t.Errorf("replica type = %q, want replicate", got)
+	}
+	if got := q.Get("ttl"); got != "15d" {
+		t.Errorf("replica ttl = %q, want the inherited volume ttl 15d", got)
+	}
+
+	select {
+	case err := <-done:
+		t.Fatalf("write returned before the replica responded: %v", err)
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		readBack := &needle.Needle{Id: n.Id}
+		if _, err := store.ReadVolumeNeedle(volumeId, readBack, nil, nil); err == nil && string(readBack.Data) == string(data) {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("local write did not land while the replica was still blocked")
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+
+	release()
+	if err := <-done; err != nil {
+		t.Fatalf("ReplicatedWrite: %v", err)
+	}
+}
+
+// TestReplicatedWriteFailsWhenEitherLegFails checks that a failed local write
+// or a failed replica fan-out fails the whole request even when the other leg
+// succeeded.
+func TestReplicatedWriteFailsWhenEitherLegFails(t *testing.T) {
+	for _, tc := range []struct {
+		name          string
+		volumeId      needle.VolumeId
+		readonlyLocal bool
+		replicaStatus int
+	}{
+		{name: "local write fails", volumeId: 8, readonlyLocal: true, replicaStatus: http.StatusCreated},
+		{name: "replica write fails", volumeId: 9, replicaStatus: http.StatusInternalServerError},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			replica := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.WriteHeader(tc.replicaStatus)
+			}))
+			defer replica.Close()
+			replicaHost := strings.TrimPrefix(replica.URL, "http://")
+
+			masterFn, dialOption := startMockMasterServer(t, &mockMasterServer{
+				locations: []*master_pb.Location{{Url: "127.0.0.1:8080"}, {Url: replicaHost}},
+			})
+
+			store := newReplicateTestStore(t, t.TempDir())
+			operation.InvalidateVolumeIdLocationCache(tc.volumeId.String())
+			if err := store.AddVolume(tc.volumeId, "", storage.NeedleMapInMemory, "001", "", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil {
+				t.Fatal(err)
+			}
+			if tc.readonlyLocal {
+				if err := store.MarkVolumeReadonly(tc.volumeId, false, false); err != nil {
+					t.Fatal(err)
+				}
+			}
+
+			data := []byte("write data")
+			n := &needle.Needle{
+				Id:       1,
+				Data:     data,
+				Checksum: needle.NewCRC(data),
+				Ttl:      needle.EMPTY_TTL,
+			}
+			r := httptest.NewRequest(http.MethodPost, fmt.Sprintf("http://127.0.0.1:8080/%d,01637037d6", tc.volumeId), nil)
+			if err := r.ParseForm(); err != nil {
+				t.Fatal(err)
+			}
+
+			if _, err := ReplicatedWrite(context.Background(), masterFn, dialOption, store, tc.volumeId, n, r, ""); err == nil {
+				t.Fatal("ReplicatedWrite succeeded despite a failed leg")
+			}
+		})
+	}
+}
+
 // TestReplicatedWriteRejectsWriteWithNoTarget verifies that a write this server
 // cannot store and has no replica to forward to fails instead of being
 // acknowledged with nothing written.

weed/util/version/constants.go

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