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 <seydi.birjand@gmail.com>

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) {

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,
 	}

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,
 	})

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,
 	})

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,
 				}

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)

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 {

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]

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 {