From 7b43dec665f18cf29a1718f8576f00f10d3a036a Mon Sep 17 00:00:00 2001 From: Junker der Provinz Date: Sat, 24 Oct 2026 19:29:38 +0200 Subject: [PATCH 01/10] shell: run volume.configure.replication in parallel per volume server (#12041) * shell: run volume.configure.replication in parallel per volume server Volume servers were configured one after another, and each server's result overwrote the previous error, so a failure on one server was lost when a later server succeeded. Run one job per volume server, bounded by -maxParallelization like the other shell commands, and return the errors from all servers. VolumeConfigure unmounts and remounts the volume, so replicas of the same volume on different servers take turns instead of all going offline at once. Co-authored-by: NyaMisty <5344431+NyaMisty@users.noreply.github.com> * shell: test that volume.configure.replication overlaps servers up to the limit * shell: assert every server ran in the parallel configure-replication test Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: NyaMisty <5344431+NyaMisty@users.noreply.github.com> Co-authored-by: chrislusf Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../command_volume_configure_replication.go | 72 ++++-- ...mmand_volume_configure_replication_test.go | 226 ++++++++++++++++++ 2 files changed, 277 insertions(+), 21 deletions(-) create mode 100644 weed/shell/command_volume_configure_replication_test.go diff --git a/weed/shell/command_volume_configure_replication.go b/weed/shell/command_volume_configure_replication.go index 8add4aad80..cc85bb5f34 100644 --- a/weed/shell/command_volume_configure_replication.go +++ b/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) { diff --git a/weed/shell/command_volume_configure_replication_test.go b/weed/shell/command_volume_configure_replication_test.go new file mode 100644 index 0000000000..f24e65b835 --- /dev/null +++ b/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) + } +} From f9dac64374cad2952c35f8159b64ba47334cedcc Mon Sep 17 00:00:00 2001 From: Junker der Provinz Date: Sat, 24 Oct 2026 19:45:57 +0200 Subject: [PATCH 02/10] helm: remove filer.redirectOnRead, which weed filer does not accept (#12040) weed filer lost -redirectOnRead in 9b3109a5d (1.59), which also removed the chart value, but 2bed15f1d added it back. Setting it makes the filer exit with "flag provided but not defined". --- k8s/charts/seaweedfs/templates/filer/filer-statefulset.yaml | 3 --- k8s/charts/seaweedfs/values.yaml | 3 --- 2 files changed, 6 deletions(-) diff --git a/k8s/charts/seaweedfs/templates/filer/filer-statefulset.yaml b/k8s/charts/seaweedfs/templates/filer/filer-statefulset.yaml index 01210f6171..04321b49bb 100644 --- a/k8s/charts/seaweedfs/templates/filer/filer-statefulset.yaml +++ b/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 }} diff --git a/k8s/charts/seaweedfs/values.yaml b/k8s/charts/seaweedfs/values.yaml index 6c6b8673db..2475590e53 100644 --- a/k8s/charts/seaweedfs/values.yaml +++ b/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 From 03f773f08728790f73a8a042d66a556a07a6b908 Mon Sep 17 00:00:00 2001 From: Junker der Provinz Date: Sun, 25 Oct 2026 00:04:37 +0200 Subject: [PATCH 03/10] redis2: pipeline entry reads when listing a directory (#12042) ListDirectoryEntries fetched every member with its own GET, so a page of 1024 entries cost 1024 round trips. Fetch the values in pipelined batches of 128 as the loop reaches them and walk the replies in order. Each reply keeps its own error, so missing and expired members are cleaned up and skipped exactly as before, and an unreadable member still ends the listing at that name. Batching keeps a caller that lists with a huge limit and stops after a few entries from reading the whole directory. Co-authored-by: Patrick Schmidt <1310072+Woellchen@users.noreply.github.com> Co-authored-by: user.email --- weed/filer/redis2/universal_redis_store.go | 36 +++++++- .../redis2/universal_redis_store_test.go | 89 +++++++++++++++++++ 2 files changed, 122 insertions(+), 3 deletions(-) diff --git a/weed/filer/redis2/universal_redis_store.go b/weed/filer/redis2/universal_redis_store.go index 1320495583..2f90230e33 100644 --- a/weed/filer/redis2/universal_redis_store.go +++ b/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)) { diff --git a/weed/filer/redis2/universal_redis_store_test.go b/weed/filer/redis2/universal_redis_store_test.go index 1558538dad..0bce68201c 100644 --- a/weed/filer/redis2/universal_redis_store_test.go +++ b/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, "") From 216c1032e6d822a50e7b98f85bb449777ddbe2d6 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 24 Oct 2026 15:06:19 -0700 Subject: [PATCH 04/10] test: mismatched If-None-Match serves the object over If-Modified-Since (#12045) The precedence case asserted the old If-Modified-Since-first ordering and expected 304. The volume servers now evaluate If-None-Match first, so a non-matching tag must return 200 with the body even when If-Modified-Since would allow 304 (RFC 9110 13.1.3). --- test/volume_server/http/read_path_variants_test.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/test/volume_server/http/read_path_variants_test.go b/test/volume_server/http/read_path_variants_test.go index eac72079a9..a997080d36 100644 --- a/test/volume_server/http/read_path_variants_test.go +++ b/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) From e442eda88e0215264b6eda8197667399efa428d6 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 24 Oct 2026 15:09:34 -0700 Subject: [PATCH 05/10] s3: authorize POST policy uploads through the shared object check (#12047) * s3: authorize POST policy uploads through the shared object check doesPolicySignatureMatch gated a verified POST Object signature on the identity's legacy Write action, which cannot represent attached IAM policies, STS session policies, or key-scoped grants. Users authorized only by an attached policy (e.g. s3:PutObject on bucket/*) were denied with AccessDenied even though the equivalent PUT succeeds. The handler already runs AuthorizeObjectWrite against the real object key after the form's x-amz-* fields are promoted to headers, so the coarse check only shadowed the full evaluation. Signature verification now just returns the identity; authorization stays in one place. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test: POST Object authorizes attached-policy identities, still denies others Cover the signature check and AuthorizeObjectWrite for an identity whose only write grant comes from an attached IAM policy, plus the deny leg for an identity whose attached policy omits s3:PutObject. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../s3api/s3api_object_handlers_postpolicy.go | 26 ++-- ...i_object_handlers_postpolicy_authz_test.go | 120 ++++++++++++++++++ 2 files changed, 128 insertions(+), 18 deletions(-) diff --git a/weed/s3api/s3api_object_handlers_postpolicy.go b/weed/s3api/s3api_object_handlers_postpolicy.go index 1e8705ba11..8ebcc71097 100644 --- a/weed/s3api/s3api_object_handlers_postpolicy.go +++ b/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) } diff --git a/weed/s3api/s3api_object_handlers_postpolicy_authz_test.go b/weed/s3api/s3api_object_handlers_postpolicy_authz_test.go index 553ff2f0a5..a0e0f7e4ce 100644 --- a/weed/s3api/s3api_object_handlers_postpolicy_authz_test.go +++ b/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") +} From 85470416cb4dd2e2b5e42a3f36612d950b5a871a Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 24 Oct 2026 15:16:16 -0700 Subject: [PATCH 06/10] volume: overlap the local needle write with the replica fan-out (#10074) * volume: overlap the local needle write with the replica fan-out ReplicatedWrite wrote the needle to local disk and then fanned it out to the replicas serially, so a replicated write cost local + replicas. The needle is already fully buffered, so the two can run at once. Snapshot the fields the fan-out reads first (the local write stamps the volume TTL and updates needle flags, so the replica goroutine must not touch the shared needle), then run the local write and the replica fan-out concurrently and ack only after both. Where local disk and replica network are separate resources the write latency drops to max(local, replicas) instead of their sum; co-located on one disk the two writes contend, so this helps distributed placements, not single-host tests. * volume: return the local isUnchanged from the write goroutine * volume: trim overlap-write comments * volume-rs: overlap the replica fan-out with the local needle write * volume: test the overlapped local write and replica fan-out * volume: decode replica pairs only when replicas exist A single-copy or incoming replicate write never uses the decoded pair map, so a malformed Pairs blob only burned an error counter and a log line. Gate the decode on replicaCount. * volume: release the stalled replica on test failure paths An early t.Fatal left the httptest handler blocked on releaseReplica, so the deferred Close hung instead of reporting the failure. Release idempotently ahead of the server close. * test: mismatched If-None-Match serves the object over If-Modified-Since The precedence case asserted the old If-Modified-Since-first ordering and expected 304. The volume servers now evaluate If-None-Match first, so a non-matching tag must return 200 with the body even when If-Modified-Since would allow 304 (RFC 9110 13.1.3). --- seaweed-volume/src/server/handlers.rs | 92 ++++++----- weed/topology/store_replicate.go | 223 +++++++++++++++----------- weed/topology/store_replicate_test.go | 173 ++++++++++++++++++++ 3 files changed, 353 insertions(+), 135 deletions(-) diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 1310793564..6826d4f3f1 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/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 { diff --git a/weed/topology/store_replicate.go b/weed/topology/store_replicate.go index fc7da99fb3..7ef9ca5694 100644 --- a/weed/topology/store_replicate.go +++ b/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 diff --git a/weed/topology/store_replicate_test.go b/weed/topology/store_replicate_test.go index 3901db744f..74eadaa48c 100644 --- a/weed/topology/store_replicate_test.go +++ b/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. From 0c32eab48e468166290c0cda76e00cc35b108b08 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Sun, 25 Oct 2026 00:48:29 +0000 Subject: [PATCH 07/10] docs: regenerate star history chart --- note/star_history.svg | 2114 ++++++++++++++++++++--------------------- 1 file changed, 1034 insertions(+), 1080 deletions(-) diff --git a/note/star_history.svg b/note/star_history.svg index 064e5b443b..72170caf1c 100644 --- a/note/star_history.svg +++ b/note/star_history.svg @@ -6,7 +6,7 @@ - 2026-10-24T02:56:41.212930 + 2026-10-25T00:48:28.860698 image/svg+xml @@ -40,23 +40,23 @@ z - + - - + - + - + - + - + - + - + - + - + - + - + @@ -301,18 +301,18 @@ L 353.368045 26.215385 - + - + - + @@ -325,18 +325,18 @@ L 449.270683 26.215385 - + - + - + @@ -349,18 +349,18 @@ L 545.042127 26.215385 - + - + - + @@ -484,41 +484,41 @@ z - + - - + - + - + - + - + - + - + - + @@ -577,18 +577,18 @@ L 709.2 267.787451 - + - + - + @@ -599,18 +599,18 @@ L 709.2 223.857989 - + - + - + @@ -621,18 +621,18 @@ L 709.2 179.928527 - + - + - + @@ -643,18 +643,18 @@ L 709.2 135.999064 - + - + - + - + - + - + @@ -813,1009 +813,963 @@ z +" clip-path="url(#p5a5d3b85f6)" style="fill: none; stroke: #0969da; stroke-width: 1.6; stroke-linecap: square"/> + From 8c1a0a0cc2979a5541a9fe41e4134dfbb41e270e Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 24 Oct 2026 19:59:19 -0700 Subject: [PATCH 08/10] filer: fix prefixed listing when the resume point sorts below the prefix (#12049) * test: prefixed directory listing with start below the prefix A listing constrained to prefix "item-b" and resumed from "item" must return item-b.versions. The seek-based stores return an empty page because the iterator lands before the prefix range and breaks on the first non-matching name. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer: clamp prefixed-listing seek to the prefix range Seek-based stores (leveldb, leveldb2, leveldb3, rocksdb, tikv, etcd) started the scan at startFileName and stopped at the first key outside the prefix. A startFileName below the prefix landed the iterator on a non-matching sibling first, so the listing returned an empty page even though matching entries followed. etcd had the mirror problem: a range beginning below the prefix end leaked non-matching entries. Begin the seek at max(startFileName, prefix), which is what foundationdb and the sql stores already do. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer(hbase): honor startFileName lower bound in prefixed listings The scan starts at the prefix but only skipped the exact startFileName, so entries sorting before it were returned again on resume. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer: keep paging a prefixed listing while matches can still follow The fallback for stores without native prefix support stopped paging as soon as the last scanned name reached the prefix, dropping prefix matches that continue on the next page. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer: return the last delivered name from the prefixed-listing fallback The fallback for stores without native prefix support pages raw entries and filters client-side. Keep the scan cursor and the returned cursor separate: resuming must continue after the last delivered match, not the last scanned name, or matches still inside the current raw page are skipped on the next page. Also keep paging while the scan cursor is inside or below the prefix range instead of stopping at the first name that reaches the prefix. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer(hbase): start prefixed-listing scans at the resume key Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/filer/etcd/etcd_store.go | 2 +- weed/filer/filerstore_wrapper.go | 65 ++++++++++------------ weed/filer/filerstore_wrapper_test.go | 38 +++++++++++++ weed/filer/hbase/hbase_store.go | 8 ++- weed/filer/leveldb/leveldb_store.go | 2 +- weed/filer/leveldb/leveldb_store_test.go | 25 +++++++++ weed/filer/leveldb2/leveldb2_store.go | 2 +- weed/filer/leveldb2/leveldb2_store_test.go | 25 +++++++++ weed/filer/leveldb3/leveldb3_store.go | 2 +- weed/filer/leveldb3/leveldb3_store_test.go | 25 +++++++++ weed/filer/rocksdb/rocksdb_store.go | 2 +- weed/filer/tikv/tikv_store.go | 2 +- 12 files changed, 154 insertions(+), 44 deletions(-) diff --git a/weed/filer/etcd/etcd_store.go b/weed/filer/etcd/etcd_store.go index 0b83bacc89..acbeb755a0 100644 --- a/weed/filer/etcd/etcd_store.go +++ b/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) } diff --git a/weed/filer/filerstore_wrapper.go b/weed/filer/filerstore_wrapper.go index 1a17e22cf6..646bd5650f 100644 --- a/weed/filer/filerstore_wrapper.go +++ b/weed/filer/filerstore_wrapper.go @@ -400,46 +400,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 diff --git a/weed/filer/filerstore_wrapper_test.go b/weed/filer/filerstore_wrapper_test.go index 0b180f63e5..6351d6978f 100644 --- a/weed/filer/filerstore_wrapper_test.go +++ b/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) +} diff --git a/weed/filer/hbase/hbase_store.go b/weed/filer/hbase/hbase_store.go index 7663fef9d3..cfa027f3c1 100644 --- a/weed/filer/hbase/hbase_store.go +++ b/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 } diff --git a/weed/filer/leveldb/leveldb_store.go b/weed/filer/leveldb/leveldb_store.go index 606166da90..341327e433 100644 --- a/weed/filer/leveldb/leveldb_store.go +++ b/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) } diff --git a/weed/filer/leveldb/leveldb_store_test.go b/weed/filer/leveldb/leveldb_store_test.go index b5384fc778..b55586ec98 100644 --- a/weed/filer/leveldb/leveldb_store_test.go +++ b/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) + } +} diff --git a/weed/filer/leveldb2/leveldb2_store.go b/weed/filer/leveldb2/leveldb2_store.go index 5c05033b99..e861db47e1 100644 --- a/weed/filer/leveldb2/leveldb2_store.go +++ b/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) } diff --git a/weed/filer/leveldb2/leveldb2_store_test.go b/weed/filer/leveldb2/leveldb2_store_test.go index 9651711d92..8d1d46ccc3 100644 --- a/weed/filer/leveldb2/leveldb2_store_test.go +++ b/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() diff --git a/weed/filer/leveldb3/leveldb3_store.go b/weed/filer/leveldb3/leveldb3_store.go index 029d902715..0b8748347a 100644 --- a/weed/filer/leveldb3/leveldb3_store.go +++ b/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) } diff --git a/weed/filer/leveldb3/leveldb3_store_test.go b/weed/filer/leveldb3/leveldb3_store_test.go index 62896a299f..a9f3be10da 100644 --- a/weed/filer/leveldb3/leveldb3_store_test.go +++ b/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) + } +} diff --git a/weed/filer/rocksdb/rocksdb_store.go b/weed/filer/rocksdb/rocksdb_store.go index 2283efa6fb..dd88d73aa3 100644 --- a/weed/filer/rocksdb/rocksdb_store.go +++ b/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) } diff --git a/weed/filer/tikv/tikv_store.go b/weed/filer/tikv/tikv_store.go index d6ba8f2844..92c34d70d6 100644 --- a/weed/filer/tikv/tikv_store.go +++ b/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) } From 1a0485a43d9b9a233d6770ea043e2e062f48a1da Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 24 Oct 2026 20:00:06 -0700 Subject: [PATCH 09/10] test: bound trino container memory + dump logs when it exits before ready (#12044) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * test: dump trino container state and logs when it exits before ready When the Trino container dies during startup, waitForTrino only reported the docker exec error ("container is not running"), which gives no hint whether the JVM was OOM-killed, the image broke, or the config is bad. Capture docker inspect exit state and the container logs in the failure message so runner-side flakes are diagnosable from the test output alone. * test: cap trino container memory at 4g in catalog tests Trino's jvm.config sizes the heap at 80% of host RAM. On a loaded self-hosted runner the JVM attempts to claim most of system memory at boot and the container is killed ~15s after docker run — the flake seen in CI today. Bounding the container to 4g gives the JVM a fixed ~3.2g ceiling (observed usage is ~750mb), immune to host memory pressure. * test: keep docker errors in trino container diagnostics Report docker inspect/logs failures instead of dropping them, and include the last docker exec output in the container-dead fatal so a missing container still explains what happened. * test: detect trino container exit in readiness fallback When the polling loop times out while the container still appears alive, the trino --version fallback may itself fail because the container has since exited. Re-check its output for the exited-container signature and route through the diagnostic dump. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test: dump trino container diagnostics on readiness timeout too The alive-but-never-ready timeout path reported only the last docker exec error, losing the container state and logs that the dead-container path already captures. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../catalog_trino/trino_catalog_test.go | 42 +++++++++++++++---- 1 file changed, 34 insertions(+), 8 deletions(-) diff --git a/test/s3tables/catalog_trino/trino_catalog_test.go b/test/s3tables/catalog_trino/trino_catalog_test.go index ddfc60fdeb..8a50318069 100644 --- a/test/s3tables/catalog_trino/trino_catalog_test.go +++ b/test/s3tables/catalog_trino/trino_catalog_test.go @@ -466,6 +466,7 @@ func (env *TestEnvironment) startTrinoContainer(t *testing.T, configDir string) cmd := exec.Command("docker", "run", "-d", "--name", containerName, + "--memory", "4g", "--add-host", "host.docker.internal:host-gateway", "-v", fmt.Sprintf("%s:/etc/trino/catalog", configDir), "-v", fmt.Sprintf("%s:/test", env.dataDir), @@ -484,7 +485,7 @@ func waitForTrino(t *testing.T, containerName string, timeout time.Duration) { deadline := time.Now().Add(timeout) var lastOutput []byte - retryCount := 0 + containerDead := false for time.Now().Before(deadline) { // Try system catalog query as a readiness check cmd := exec.Command("docker", "exec", containerName, @@ -495,26 +496,51 @@ 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 } - retryCount++ } time.Sleep(1 * time.Second) } // 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) } - t.Fatalf("Timed out waiting for Trino to be ready\nLast output:\n%s", string(lastOutput)) + 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\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 { + var b strings.Builder + if out, err := exec.Command("docker", "inspect", "-f", + "status={{.State.Status}} exitCode={{.State.ExitCode}} oomKilled={{.State.OOMKilled}} error={{.State.Error}}", + containerName).CombinedOutput(); err == nil { + fmt.Fprintf(&b, "docker inspect: %s\n", strings.TrimSpace(string(out))) + } else { + fmt.Fprintf(&b, "docker inspect failed: %v (%s)\n", err, strings.TrimSpace(string(out))) + } + if out, err := exec.Command("docker", "logs", "--tail", "80", containerName).CombinedOutput(); err == nil { + fmt.Fprintf(&b, "docker logs:\n%s", string(out)) + } else { + fmt.Fprintf(&b, "docker logs failed: %v (%s)", err, strings.TrimSpace(string(out))) + } + return b.String() } func runTrinoSQL(t *testing.T, containerName, sql string) string { From ec8f9793502c59ce3381cc4151bfc0806afd92e8 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Sun, 25 Oct 2026 03:09:42 +0000 Subject: [PATCH 10/10] 4.50 --- k8s/charts/seaweedfs/Chart.yaml | 4 ++-- weed/util/version/constants.go | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/k8s/charts/seaweedfs/Chart.yaml b/k8s/charts/seaweedfs/Chart.yaml index 247b5bd06c..5248f746de 100644 --- a/k8s/charts/seaweedfs/Chart.yaml +++ b/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: 4.49.0 +version: 4.50.0 diff --git a/weed/util/version/constants.go b/weed/util/version/constants.go index 852d163e40..37a1520405 100644 --- a/weed/util/version/constants.go +++ b/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 = ""