s3api: push the listing prefix down to the filer in ListObjectVersions (#11070)

The version walk listed every directory with no prefix, transferring
all 1024-entry batches over gRPC and filtering gateway-side - and kept
paging past the point where names can no longer match. On wide
directories (many sibling orgs/jobs next to the requested prefix) that
is most of the transfer, decode, and CPU cost of every page.

Derive the next path component of the requested prefix per directory
level and hand it to the filer listing. A name holds no slash, so a
directory whose name does not start with the component cannot contain
a matching key and a file that does not cannot be one; stores with
native prefixed listing (sql, leveldb) turn this into a range scan and
stop the stream at the end of the prefix zone.

Claude-Session: https://claude.ai/code/session_01FquvGtTD2zA3uMZGQHAuV4

weed/s3api/s3api_object_versioning.go

@@ -550,6 +550,37 @@ func (vc *versionCollector) computeStartFrom(relativePath string) (startFrom str
 	return remainder, true
 }
 
+// computeListPrefix returns the name prefix that every entry of the directory
+// at relativePath must carry to be relevant to vc.prefix: the next path
+// component of the requested prefix under that directory. Pushing it into the
+// filer listing stops unrelated siblings from being transferred and scanned at
+// every level on the way down to the prefix. A name can hold no slash, so a
+// directory whose name does not start with this component cannot contain a
+// matching key, and a file whose name does not start with it cannot be one.
+// Inside the prefix zone (and without a prefix) it returns "" - no constraint.
+func (vc *versionCollector) computeListPrefix(relativePath string) string {
+	if vc.prefix == "" {
+		return ""
+	}
+
+	var remainder string
+	if relativePath == "" {
+		remainder = vc.prefix
+	} else if strings.HasPrefix(vc.prefix, relativePath+"/") {
+		remainder = vc.prefix[len(relativePath)+1:]
+	} else {
+		// The walk only enters a directory that matches the prefix or can
+		// descend toward it; one the prefix does not extend into means the
+		// whole directory is inside the prefix zone.
+		return ""
+	}
+
+	if idx := strings.Index(remainder, "/"); idx >= 0 {
+		return remainder[:idx]
+	}
+	return remainder
+}
+
 // shouldSkipObjectForMarker returns true if the object should be skipped based on keyMarker
 func (vc *versionCollector) shouldSkipObjectForMarker(objectKey string) bool {
 	if vc.keyMarker == "" {
@@ -795,12 +826,13 @@ func (vc *versionCollector) collectVersions(currentPath, relativePath string) er
 		startFrom = markerStart
 		inclusive = true
 	}
+	listPrefix := vc.computeListPrefix(relativePath)
 	for {
 		if vc.isFull() {
 			return nil
 		}
 
-		entries, isLast, err := vc.s3a.list(currentPath, "", startFrom, inclusive, filer.PaginationSize)
+		entries, isLast, err := vc.s3a.list(currentPath, listPrefix, startFrom, inclusive, filer.PaginationSize)
 		// After the first batch, use exclusive mode for standard pagination
 		inclusive = false
 		if err != nil {

weed/s3api/s3api_object_versioning_list_prefix_test.go

@@ -0,0 +1,41 @@
+package s3api
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// The component handed to the filer must keep every entry that can match or
+// descend toward the requested prefix at that directory level, while excluding
+// siblings that cannot: a name holds no slash, so a directory not starting
+// with the component cannot contain a matching key.
+func TestComputeListPrefix(t *testing.T) {
+	cases := []struct {
+		prefix       string
+		relativePath string
+		want         string
+	}{
+		{"", "", ""},
+		{"", "a/b", ""},
+		{"a/b/c", "", "a"},
+		{"a/b/c", "a", "b"},
+		{"a/b/c", "a/b", "c"},
+		// at or inside the prefix zone: no constraint
+		{"a/b/c", "a/b/c", ""},
+		{"a/b/", "a/b", ""},
+		{"a", "a/sub", ""},
+		// partial component narrows the listing but keeps deeper matches
+		// ("a/bc" keeps dir "bc", file "bcd", and "bc.versions" at level "a")
+		{"a/bc", "a", "bc"},
+		{"abc", "", "abc"},
+		// a directory off the prefix path is inside the zone by construction
+		{"a/b/c", "x", ""},
+	}
+
+	for _, tc := range cases {
+		vc := &versionCollector{prefix: tc.prefix}
+		assert.Equal(t, tc.want, vc.computeListPrefix(tc.relativePath),
+			"prefix %q at %q", tc.prefix, tc.relativePath)
+	}
+}