From 3dd53eb81f20cc0525176dcf2c7006dc1c12dcba Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sat, 17 Oct 2026 21:23:28 +0800 Subject: [PATCH] fix(listing): preserve directories after absent object recovery Signed-off-by: Feng Ruohang (cherry picked from commit 1445fc1d85d5166de24c7a378cee3479cfc95a43) Signed-off-by: Feng Ruohang --- cmd/metacache-entries.go | 26 +++- cmd/metacache-listing-directory_test.go | 181 ++++++++++++++++++++++++ 2 files changed, 203 insertions(+), 4 deletions(-) create mode 100644 cmd/metacache-listing-directory_test.go diff --git a/cmd/metacache-entries.go b/cmd/metacache-entries.go index 3a0600bee..ee45ac492 100644 --- a/cmd/metacache-entries.go +++ b/cmd/metacache-entries.go @@ -781,6 +781,15 @@ func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntr } last := "" var toMerge []int + var directoryCollisions []int + discardDirectories := func() error { + for _, idx := range directoryCollisions { + if err := selectFrom(idx); err != nil { + return err + } + } + return nil + } // Choose the best to return. for { @@ -790,6 +799,7 @@ func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntr best := top[0] bestIdx := 0 toMerge = toMerge[:0] + directoryCollisions = directoryCollisions[:0] for i, other := range top[1:] { otherIdx := i + 1 if other == nil { @@ -819,10 +829,9 @@ func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntr if serverDebugLog { console.Debugln("mergeEntryChannels: discarding directory", other.name, "for object", best.name) } - // Discard the directory. - if err := selectFrom(otherIdx); err != nil { - return err - } + // A pending object can resolve to absence. Keep the + // directory until the whole object group is confirmed. + directoryCollisions = append(directoryCollisions, otherIdx) continue } // Replace directory with object. @@ -838,6 +847,7 @@ func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntr } if best.name > other.name { toMerge = toMerge[:0] + directoryCollisions = directoryCollisions[:0] best = other bestIdx = otherIdx } @@ -866,6 +876,11 @@ func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntr if err != nil { return err } + if resolved != nil { + if err := discardDirectories(); err != nil { + return err + } + } if resolved != nil && resolved.name > last { select { case <-ctxDone: @@ -876,6 +891,9 @@ func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntr } continue } + if err := discardDirectories(); err != nil { + return err + } // Merge any unmerged if len(toMerge) > 0 { diff --git a/cmd/metacache-listing-directory_test.go b/cmd/metacache-listing-directory_test.go new file mode 100644 index 000000000..6718f08f2 --- /dev/null +++ b/cmd/metacache-listing-directory_test.go @@ -0,0 +1,181 @@ +// Copyright (c) 2026 Ruohang Feng +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cmd + +import ( + "bytes" + "context" + "encoding/xml" + "fmt" + "io" + "net/http" + "net/http/httptest" + "slices" + "testing" + "time" +) + +type listingDirectoryWalkSnapshot struct { + StorageAPI + bucket string + snapshot []byte +} + +func (d listingDirectoryWalkSnapshot) WalkDir(ctx context.Context, opts WalkDirOptions, out io.Writer) error { + if opts.Bucket != d.bucket { + return d.StorageAPI.WalkDir(ctx, opts, out) + } + _, err := out.Write(d.snapshot) + return err +} + +// Capture two actual directory-walk generations for a directory object, then +// acknowledge its deletion. A different pool still contains dir/child. The +// stale source walkers disagree, while every current source drive is absent. +func TestListingRecoveryDeletedDirectoryKeepsCommonPrefix(t *testing.T) { + for _, marker := range []bool{false, true} { + for _, sourcePool := range []int{0, 1} { + t.Run(fmt.Sprintf("pool=%d/marker=%t", sourcePool, marker), func(t *testing.T) { + z, bucket, router := listingPoolsBackend(t) + ctx := t.Context() + source := z.serverPools[sourcePool].sets[0] + target := z.serverPools[1-sourcePool].sets[0] + childBody := []byte("surviving child") + put := func(set *erasureObjects, key string, tm time.Time) { + var body []byte + if key == "dir/child" { + body = childBody + } + _, err := set.PutObject(ctx, bucket, encodeDirObject(key), mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{MTime: tm, VersionSuspended: marker}) + if err != nil { + t.Fatal(err) + } + } + tm := time.Now().UTC().Add(-time.Hour) + put(target, "dir/child", tm) + disks, switchDisks := installListingDiskSwitch(source) + snapshots := make([][]byte, len(disks)) + for gen := range 3 { + if gen < 2 { + put(source, "dir/", tm.Add(time.Duration(gen)*time.Minute)) + } else { + if _, err := source.DeleteObject(ctx, bucket, encodeDirObject("dir/"), ObjectOptions{VersionSuspended: marker, DeleteMarker: marker, MTime: tm.Add(2 * time.Minute)}); err != nil { + t.Fatal(err) + } + } + for i, disk := range disks { + wanted := 0 + if i >= 6 { + wanted = 1 + } + if i >= 12 { + wanted = 2 + } + if wanted != gen { + continue + } + var b bytes.Buffer + if err := disk.WalkDir(ctx, WalkDirOptions{Bucket: bucket, Recursive: false}, &b); err != nil { + t.Fatal(err) + } + snapshots[i] = append([]byte(nil), b.Bytes()...) + } + } + for _, key := range []string{"dir/child", "dir/"} { + request, err := newTestSignedRequestV4(http.MethodGet, getGetObjectURL("", bucket, key), 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil) + if err != nil { + t.Fatal(err) + } + response := httptest.NewRecorder() + router.ServeHTTP(response, request.WithContext(ctx)) + t.Logf("GET %s HTTP %d", key, response.Code) + want := http.StatusOK + if key == "dir/" { + want = http.StatusNotFound + } + if response.Code != want { + t.Fatalf("GET %s: got %d want %d: %s", key, response.Code, want, response.Body.String()) + } + if key == "dir/child" && !bytes.Equal(response.Body.Bytes(), childBody) { + t.Fatal("surviving child content changed") + } + } + wrapped := make([]StorageAPI, len(disks)) + for i, disk := range disks { + wrapped[i] = listingDirectoryWalkSnapshot{disk, bucket, snapshots[i]} + } + switchDisks(wrapped) + t.Cleanup(func() { switchDisks(disks) }) + req, err := newTestSignedRequestV4(http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", "/"), 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req.WithContext(ctx)) + t.Logf("LIST HTTP %d body=%s", rec.Code, rec.Body.String()) + var page ListObjectsV2Response + if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &page) != nil { + t.Fatalf("LIST failed: %d %s", rec.Code, rec.Body.String()) + } + if len(page.CommonPrefixes) != 1 || page.CommonPrefixes[0].Prefix != "dir/" { + t.Fatalf("acknowledged child lost its CommonPrefix: %+v", page) + } + }) + } + } +} + +// The directory must remain in the sorted input when an earlier object name +// disappears. Emitting it immediately would skip an intervening name, and +// keeping only one colliding directory would leave input order significant. +func TestListingRecoveryDirectoryCollisionOrdering(t *testing.T) { + for _, object := range []string{"dir", "dir/"} { + for _, present := range []bool{false, true} { + if present && object == "dir" { + continue + } + for source := range 3 { + t.Run(fmt.Sprintf("object=%s/present=%t/source=%d", object, present, source), func(t *testing.T) { + in := make([]chan metaCacheEntry, 6) + for i := range in { + in[i] = make(chan metaCacheEntry, 1) + entry := metaCacheEntry{name: "dir/"} + if i == source { + entry = metaCacheEntry{name: object, listingSet: &erasureObjects{}} + } + if i >= 3 { + entry = metaCacheEntry{name: []string{"dir-", "dir.ext", "dir0"}[i-3], metadata: []byte{1}} + } + in[i] <- entry + close(in[i]) + } + out := make(chan metaCacheEntry, len(in)) + err := mergeEntryChannelsWithResolver(t.Context(), in, out, 1, nil, func(group []*metaCacheEntry) (*metaCacheEntry, error) { + if len(group) != 1 || group[0].listingSet == nil { + t.Fatalf("directory entered the object group: %+v", group) + } + if !present { + return nil, nil + } + return &metaCacheEntry{name: object, metadata: []byte{1}}, nil + }) + if err != nil { + t.Fatal(err) + } + var names []string + for entry := range out { + names = append(names, entry.name) + if entry.name == "dir/" && entry.isObject() != present { + t.Fatalf("directory collision returned the wrong entry type: %+v", entry) + } + } + want := []string{"dir-", "dir.ext", "dir/", "dir0"} + if !slices.Equal(names, want) { + t.Fatalf("names=%v, want %v", names, want) + } + }) + } + } + } +}