From 258b167b1c65f1a4fa4d85dbef7ae8ef9fe4f3f6 Mon Sep 17 00:00:00 2001 From: mr javad seydi Date: Sat, 19 Sep 2026 22:11:10 +0330 Subject: [PATCH] fix(listing): resolve undecidable list entries against the locked read path Refs: https://github.com/pgsty/silo/issues/218 Signed-off-by: mr javad seydi --- cmd/data-scanner.go | 9 +- cmd/erasure-healing.go | 6 +- cmd/erasure-server-pool-decom.go | 3 +- cmd/erasure-server-pool-rebalance.go | 3 +- cmd/erasure-server-pool.go | 10 +- cmd/global-heal.go | 3 +- cmd/metacache-entries.go | 10 ++ cmd/metacache-null-quorum_test.go | 164 +++++++++++++++++++++++++++ cmd/metacache-set.go | 82 +++++++++++++- 9 files changed, 272 insertions(+), 18 deletions(-) diff --git a/cmd/data-scanner.go b/cmd/data-scanner.go index 0475c4a53..8bbfd8f31 100644 --- a/cmd/data-scanner.go +++ b/cmd/data-scanner.go @@ -744,10 +744,10 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int } }, // Some disks have data for this. - partial: func(entries metaCacheEntries, errs []error) { + partial: func(entries metaCacheEntries, errs []error) error { if !f.shouldHeal() { cancel() - return + return nil } entry, ok := entries.resolve(&resolver) if !ok { @@ -772,7 +772,7 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int } if entry.isDir() { - return + return nil } // We got an entry which we should be able to heal. @@ -787,7 +787,7 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int scannerLogIf(ctx, err) } foundObjs = foundObjs || err == nil - return + return nil } custom["versions"] = fmt.Sprint(len(fiv.Versions)) @@ -814,6 +814,7 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int } custom["success_versions"] = fmt.Sprint(successVersions) custom["failed_versions"] = fmt.Sprint(failVersions) + return nil }, // Too many disks failed. finished: func(errs []error) { diff --git a/cmd/erasure-healing.go b/cmd/erasure-healing.go index 8fb371270..e02a075f1 100644 --- a/cmd/erasure-healing.go +++ b/cmd/erasure-healing.go @@ -86,7 +86,7 @@ func (er erasureObjects) listAndHeal(ctx context.Context, bucket, prefix string, cancel() } }, - partial: func(entries metaCacheEntries, _ []error) { + partial: func(entries metaCacheEntries, _ []error) error { entry, ok := entries.resolve(&resolver) if !ok { // check if we can get one entry at least @@ -94,12 +94,12 @@ func (er erasureObjects) listAndHeal(ctx context.Context, bucket, prefix string, entry, _ = entries.firstFound() } if !recursive && prefix != entry.name { - return + return nil } if err := healEntry(bucket, *entry, scanMode); err != nil { cancel() - return } + return nil }, finished: nil, } diff --git a/cmd/erasure-server-pool-decom.go b/cmd/erasure-server-pool-decom.go index 0e422c3ae..abae13c0d 100644 --- a/cmd/erasure-server-pool-decom.go +++ b/cmd/erasure-server-pool-decom.go @@ -730,11 +730,12 @@ func (set *erasureObjects) listObjectsToDecommission(ctx context.Context, bi dec minDisks: listingQuorum, reportNotFound: false, agreed: fn, - partial: func(entries metaCacheEntries, _ []error) { + partial: func(entries metaCacheEntries, _ []error) error { entry, ok := entries.resolve(&resolver) if ok { fn(*entry) } + return nil }, finished: nil, }) diff --git a/cmd/erasure-server-pool-rebalance.go b/cmd/erasure-server-pool-rebalance.go index b60c9c904..f352c5f93 100644 --- a/cmd/erasure-server-pool-rebalance.go +++ b/cmd/erasure-server-pool-rebalance.go @@ -525,11 +525,12 @@ func (set *erasureObjects) listObjectsToRebalance(ctx context.Context, bucketNam minDisks: listingQuorum, reportNotFound: false, agreed: fn, - partial: func(entries metaCacheEntries, _ []error) { + partial: func(entries metaCacheEntries, _ []error) error { entry, ok := entries.resolve(&resolver) if ok { fn(*entry) } + return nil }, finished: nil, }) diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index df1d2f22d..b04e57589 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -2642,11 +2642,15 @@ func (z *erasureServerPools) Walk(ctx context.Context, bucket, prefix string, re minDisks: listingQuorum, reportNotFound: false, agreed: send, - partial: func(entries metaCacheEntries, _ []error) { - entry, ok := entries.resolve(&resolver) - if ok { + partial: func(entries metaCacheEntries, _ []error) error { + entry, err := set.resolveListEntry(ctx, bucket, entries, &resolver) + if err != nil { + return err + } + if entry != nil { send(*entry) } + return nil }, finished: nil, } diff --git a/cmd/global-heal.go b/cmd/global-heal.go index ca78df818..e153418c1 100644 --- a/cmd/global-heal.go +++ b/cmd/global-heal.go @@ -529,7 +529,7 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string, jt.Take() go healEntry(bucket, entry) }, - partial: func(entries metaCacheEntries, _ []error) { + partial: func(entries metaCacheEntries, _ []error) error { entry, ok := entries.resolve(&resolver) if !ok { // check if we can get one entry at least @@ -538,6 +538,7 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string, } jt.Take() go healEntry(bucket, *entry) + return nil }, finished: func(errs []error) { success := countErrs(errs, nil) diff --git a/cmd/metacache-entries.go b/cmd/metacache-entries.go index fceec0b43..166176596 100644 --- a/cmd/metacache-entries.go +++ b/cmd/metacache-entries.go @@ -461,6 +461,16 @@ func (m metaCacheEntries) firstFound() (first *metaCacheEntry, n int) { return first, n } +// firstObject returns the first entry that contains object metadata. +func (m metaCacheEntries) firstObject() *metaCacheEntry { + for i := range m { + if m[i].isObject() { + return &m[i] + } + } + return nil +} + // names will return all names in order. // Since this allocates it should not be used in critical functions. func (m metaCacheEntries) names() []string { diff --git a/cmd/metacache-null-quorum_test.go b/cmd/metacache-null-quorum_test.go index a0aa09308..8804f15f3 100644 --- a/cmd/metacache-null-quorum_test.go +++ b/cmd/metacache-null-quorum_test.go @@ -52,6 +52,14 @@ type nullQuorumWalkDisk struct { index int } +type failedMetadataWalkDisk struct { + StorageAPI +} + +func (d *failedMetadataWalkDisk) WalkDir(context.Context, WalkDirOptions, io.Writer) error { + return errDiskNotFound +} + func (d *nullQuorumWalkDisk) WalkDir(ctx context.Context, opts WalkDirOptions, out io.Writer) error { if opts.Bucket != d.schedule.bucket { return d.StorageAPI.WalkDir(ctx, opts, out) @@ -227,6 +235,162 @@ func TestListObjectsSingleNullQuorumHTTP(t *testing.T) { } } +func TestListObjectsFallsBackToReadableMetadata(t *testing.T) { + z, bucket, router := nullQuorumBackend(t) + set := z.serverPools[0].sets[0] + disks := set.getDisks() + const object = "object" + oldBody := bytes.Repeat([]byte{'a'}, 8192) + newBody := bytes.Repeat([]byte{'b'}, 8192) + oldTime := time.Now().UTC().Add(-time.Hour) + + if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(oldBody), int64(len(oldBody)), "", ""), ObjectOptions{MTime: oldTime}); err != nil { + t.Fatal(err) + } + oldMeta := make([][]byte, len(disks)) + for i, disk := range disks { + oldMeta[i] = mustReadNullQuorumMeta(t, disk, bucket, object) + } + if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(newBody), int64(len(newBody)), "", ""), ObjectOptions{MTime: oldTime.Add(time.Minute)}); err != nil { + t.Fatal(err) + } + + // The eight readable walk streams are split 3/5 across two generations; + // the other eight streams fail. Neither generation is proven readable by + // the listing inputs, so returning HTTP 200 would silently omit the key. + for i := range 3 { + if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, oldMeta[i]); err != nil { + t.Fatal(err) + } + } + wrapped := make([]StorageAPI, len(disks)) + for i, disk := range disks { + wrapped[i] = disk + if i >= 8 { + wrapped[i] = &failedMetadataWalkDisk{StorageAPI: disk} + } + } + set.getDisks = func() []StorageAPI { return wrapped } + + rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", "")) + var list ListObjectsV2Response + if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil { + t.Fatalf("LIST failed to resolve readable metadata: %d %s", rec.Code, rec.Body.String()) + } + if list.KeyCount != 1 || len(list.Contents) != 1 || list.Contents[0].Key != object { + t.Fatalf("LIST omitted the readable key: %+v", list) + } + + // The object itself still has a read quorum outside the failed walk + // streams, demonstrating why the LIST must report uncertainty, not absence. + get := nullQuorumRequest(t, router, http.MethodGet, getGetObjectURL("", bucket, object)) + if get.Code != http.StatusOK || !bytes.Equal(get.Body.Bytes(), newBody) { + t.Fatalf("GET lost the readable generation: %d %q", get.Code, get.Body.String()) + } +} + +// The fallback must read under the object read lock. A lock-free read of an +// object being overwritten stably reports a live object as absent, which is +// the omission this repair exists to remove. Hold the write lock and assert +// the listing waits for it rather than resolving straight through. +func TestListFallbackTakesObjectReadLock(t *testing.T) { + z, bucket, router := nullQuorumBackend(t) + set := z.serverPools[0].sets[0] + disks := set.getDisks() + const object = "object" + oldBody, newBody := bytes.Repeat([]byte{'a'}, 8192), bytes.Repeat([]byte{'b'}, 8192) + oldTime := time.Now().UTC().Add(-time.Hour) + + if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(oldBody), int64(len(oldBody)), "", ""), ObjectOptions{MTime: oldTime}); err != nil { + t.Fatal(err) + } + oldMeta := make([][]byte, len(disks)) + for i, disk := range disks { + oldMeta[i] = mustReadNullQuorumMeta(t, disk, bucket, object) + } + if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(newBody), int64(len(newBody)), "", ""), ObjectOptions{MTime: oldTime.Add(time.Minute)}); err != nil { + t.Fatal(err) + } + // Split the readable walk streams so the entry cannot be resolved from + // the snapshots and the fallback has to run. + for i := range 3 { + if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, oldMeta[i]); err != nil { + t.Fatal(err) + } + } + wrapped := make([]StorageAPI, len(disks)) + for i, disk := range disks { + wrapped[i] = disk + if i >= 8 { + wrapped[i] = &failedMetadataWalkDisk{StorageAPI: disk} + } + } + set.getDisks = func() []StorageAPI { return wrapped } + + lock := set.NewNSLock(bucket, object) + lkctx, err := lock.GetLock(t.Context(), globalOperationTimeout) + if err != nil { + t.Fatal(err) + } + const hold = 300 * time.Millisecond + released := make(chan struct{}) + go func() { + time.Sleep(hold) + lock.Unlock(lkctx) + close(released) + }() + + start := time.Now() + rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", "")) + elapsed := time.Since(start) + <-released + + if elapsed < hold { + t.Fatalf("listing fallback did not wait for the object lock: %v < %v", elapsed, hold) + } + var list ListObjectsV2Response + if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil { + t.Fatalf("LIST failed after the lock was released: %d %s", rec.Code, rec.Body.String()) + } + if list.KeyCount != 1 || len(list.Contents) != 1 || list.Contents[0].Key != object { + t.Fatalf("LIST omitted the readable key: %+v", list) + } +} + +// A truncated xl.meta is what a concurrent overwrite looks like on one drive, +// and the walker skips that entry. Skipping is safe and deliberate: the key is +// only omitted when no generation reaches quorum, which listPath now resolves +// against the read path. Pin that, so the walker is not "hardened" into +// failing a whole listing over a minority of unreadable drives. +func TestListObjectsSurvivesTruncatedMetadataMinority(t *testing.T) { + z, bucket, router := nullQuorumBackend(t) + set := z.serverPools[0].sets[0] + disks := set.getDisks() + const object = "object" + body := bytes.Repeat([]byte{'a'}, 8192) + + if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{}); err != nil { + t.Fatal(err) + } + + // Truncate xl.meta on a minority of drives, as an overwrite in flight does. + for i := range 4 { + meta := mustReadNullQuorumMeta(t, disks[i], bucket, object) + if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, meta[:len(meta)/2]); err != nil { + t.Fatal(err) + } + } + + rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", "")) + var list ListObjectsV2Response + if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil { + t.Fatalf("a truncated minority failed the listing: %d %s", rec.Code, rec.Body.String()) + } + if list.KeyCount != 1 || len(list.Contents) != 1 || list.Contents[0].Key != object { + t.Fatalf("a truncated minority omitted the key: %+v", list) + } +} + func TestSingleNullQuorumReadAndConsumers(t *testing.T) { z, bucket, router := nullQuorumBackend(t) set := z.serverPools[0].sets[0] diff --git a/cmd/metacache-set.go b/cmd/metacache-set.go index a2ab603c6..16eb022a0 100644 --- a/cmd/metacache-set.go +++ b/cmd/metacache-set.go @@ -700,6 +700,72 @@ func getQuorumDisks(disks []StorageAPI, infos []DiskInfo, readQuorum int) (newDi return newDisks } +// resolveListEntry resolves walk snapshots and, for latest-only listings, +// falls back to the regular object read path when those snapshots disagree. +// +// Cost: one locked getObjectFileInfo per unresolved entry. Only disagreeing +// entries pay it -- 123 of ~162k entries in the issue #218 rolling-restart run +// -- but each is a namespace lock round trip, so a listing taken during a +// restart of a large bucket can pay it per page. Batch the fallback reads if +// listing latency during restarts becomes the complaint. +func (er *erasureObjects) resolveListEntry(ctx context.Context, bucket string, entries metaCacheEntries, resolver *metadataResolutionParams) (*metaCacheEntry, error) { + entry, ok := entries.resolve(resolver) + if ok { + return entry, nil + } + entry = entries.firstObject() + if entry == nil { + return nil, nil + } + readQuorumErr := func() error { + return InsufficientReadQuorum{ + Bucket: bucket, + Object: entry.name, + Err: errErasureReadQuorum, + Type: RQInconsistentMeta, + } + } + if resolver.requestedVersions != 1 { + return nil, readQuorumErr() + } + + // Read under the object read lock, exactly as the GET path does. A + // lock-free read of an object being overwritten stably reports a live + // object as absent, so an unlocked fallback only trades one silent + // omission for another. + lock := er.NewNSLock(bucket, entry.name) + lkctx, err := lock.GetRLock(ctx, globalOperationTimeout) + if err != nil { + return nil, err + } + defer lock.RUnlock(lkctx) + + fi, _, _, err := er.getObjectFileInfo(lkctx.Context(), bucket, entry.name, ObjectOptions{}, false) + if err != nil { + // Absence is reported only when the locked read proves it. Anything + // else is undecidable and fails the request instead of dropping the + // key from a successful listing. + if isErrObjectNotFound(err) || isErrVersionNotFound(err) { + return nil, nil + } + return nil, err + } + if fi.Deleted { + return nil, nil + } + + xl := &xlMetaV2{} + if err = xl.AddVersion(fi); err != nil { + return nil, readQuorumErr() + } + resolved := &metaCacheEntry{name: entry.name, cached: xl, reusable: true} + resolved.metadata, err = xl.AppendTo(metaDataPoolGet()) + if err != nil { + return nil, err + } + return resolved, nil +} + // Will return io.EOF if continuing would not yield more results. func (er *erasureObjects) listPath(ctx context.Context, o listPathOptions, results chan<- metaCacheEntry) (err error) { defer xioutil.SafeClose(results) @@ -776,15 +842,19 @@ func (er *erasureObjects) listPath(ctx context.Context, o listPathOptions, resul case results <- entry: } }, - partial: func(entries metaCacheEntries, errs []error) { + partial: func(entries metaCacheEntries, _ []error) error { // Results Disagree :-( - entry, ok := entries.resolve(&resolver) - if ok { + entry, err := er.resolveListEntry(ctx, o.Bucket, entries, &resolver) + if err != nil { + return err + } + if entry != nil { select { case <-ctxDone: case results <- *entry: } } + return nil }, }) } @@ -980,7 +1050,7 @@ type listPathRawOptions struct { // partial will be called when there is disagreement between disks. // if disk did not return any result, but also haven't errored // the entry will be empty and errs will - partial func(entries metaCacheEntries, errs []error) + partial func(entries metaCacheEntries, errs []error) error // finished will be called when all streams have finished and // more than one disk returned an error. @@ -1191,7 +1261,9 @@ func listPathRaw(ctx context.Context, opts listPathRawOptions) (err error) { continue } if opts.partial != nil { - opts.partial(topEntries, errs) + if err := opts.partial(topEntries, errs); err != nil { + return err + } } // Skip the inputs we used. for i, r := range readers {