s3: list children of the directory a ListObjects marker names (#11805)

* s3: resume into the directory a listing marker names

* s3: keep delimited marker resumes exclusive

* s3: descend a marker-named directory without repeating its key

* s3: emit marker-named dir objects but keep internal folders hidden

weed/s3api/s3api_object_handlers_list.go

@@ -950,6 +950,10 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.
 		return
 	}
 
+	// markerParentWalked tracks whether marker now names a directory whose
+	// subtree this call already walked: after the descent below it is the
+	// parent directory, and seeing its entry again must not re-list it.
+	markerParentWalked := false
 	if strings.Contains(marker, "/") {
 		subDir, subMarker := toParentAndDescendants(marker)
 		// println("doListFilerEntries dir", dir+"/"+subDir, "subMarker", subMarker)
@@ -961,11 +965,21 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.
 		nextMarker = subDir + "/" + subNextMarker
 		// finished processing this subdirectory
 		marker = subDir
+		markerParentWalked = true
 	}
 	if cursor.isTruncated {
 		return
 	}
 
+	// Ask for the marker's own entry: an exclusive window never delivers the
+	// directory a marker names, and its children sort after the marker key.
+	// Delimited listings keep exclusive windows — a marker equal to a
+	// directory there is the resume position of an already-emitted common
+	// prefix, not a subtree to walk.
+	if marker != "" && delimiter == "" {
+		inclusiveStartFrom = true
+	}
+
 	// now marker is also a direct child of dir
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
@@ -1023,7 +1037,37 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.
 			if entry.IsDirectory {
 				markerName = strings.TrimSuffix(markerName, s3_constants.VersionsFolder)
 			}
-			if !inclusiveStartFrom && marker != "" && (entry.Name == marker || markerName == marker) {
+			if marker != "" && (entry.Name == marker || markerName == marker) {
+				// The marker excludes only the key it names. A directory under
+				// that name sorts its children after the marker, so a recursive
+				// listing still walks the subtree — from the start. A directory
+				// key object lists as "<marker>/" — a different key that also
+				// sorts after — so it is emitted; a prefix object would repeat
+				// the marker itself and is not. Internal folders keep their
+				// exclusion, .versions dirs fold into the marker key, and a
+				// delimited listing keeps skipping — the same marker resumes an
+				// already-emitted common prefix there.
+				walkableMarkerDir := delimiter == "" && entry.IsDirectory && entry.Name == marker &&
+					!markerParentWalked && !strings.HasSuffix(entry.Name, s3_constants.VersionsFolder) &&
+					entry.Name != s3_constants.MultipartUploadsFolder
+				if !walkableMarkerDir {
+					continue
+				}
+				if entry.IsDirectoryKeyObject() && !entry.IsPrefixObject() {
+					eachEntryFn(dir, entry)
+					if cursor.isTruncated {
+						return
+					}
+				}
+				subNextMarker, subErr := s3a.doListFilerEntries(ctx, client, listDirectoryRequest{dir: dir + "/" + entry.Name, delimiter: delimiter, bucket: bucket}, cursor, eachEntryFn)
+				if subErr != nil {
+					err = fmt.Errorf("doListFilerEntries: %w", subErr)
+					return
+				}
+				nextMarker = entry.Name + "/" + subNextMarker
+				if cursor.isTruncated {
+					return
+				}
 				continue
 			}
 

weed/s3api/s3api_object_handlers_list_test.go

@@ -423,6 +423,50 @@ func TestDoListFilerEntries_ExclusiveStartSkipsMarkerEchoWithSubsequentEntries(t
 	assert.Equal(t, "zebra.txt", nextMarker)
 }
 
+func TestDoListFilerEntries_MarkerNamingDirectoryListsItsChildren(t *testing.T) {
+	// StartAfter=d00/s01 must still list d00/s01's contents: every child sorts
+	// after the marker key. The same holds for a bare top-level directory.
+	s3a := &S3ApiServer{}
+	dir := func(name string) *filer_pb.Entry {
+		return &filer_pb.Entry{Name: name, IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}}
+	}
+	file := func(name string) *filer_pb.Entry {
+		return &filer_pb.Entry{Name: name, Attributes: &filer_pb.FuseAttributes{}}
+	}
+	client := &testFilerClient{
+		entriesByDir: map[string][]*filer_pb.Entry{
+			"/buckets/test-bucket":         {dir("d00"), dir("d01")},
+			"/buckets/test-bucket/d00":     {dir("s00"), dir("s01"), dir("s02")},
+			"/buckets/test-bucket/d00/s00": {file("f000.txt")},
+			"/buckets/test-bucket/d00/s01": {file("f000.txt")},
+			"/buckets/test-bucket/d00/s02": {file("f000.txt")},
+			"/buckets/test-bucket/d01":     {dir("s00")},
+			"/buckets/test-bucket/d01/s00": {file("f000.txt")},
+		},
+	}
+
+	collect := func(marker string) []string {
+		cursor := &ListingCursor{maxKeys: 1000}
+		var seen []string
+		_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test-bucket", marker: marker, bucket: "test-bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
+			seen = append(seen, dir+"/"+entry.Name)
+			cursor.maxKeys--
+		})
+		assert.NoError(t, err)
+		return seen
+	}
+
+	assert.Equal(t, []string{
+		"/buckets/test-bucket/d00/s01/f000.txt",
+		"/buckets/test-bucket/d00/s02/f000.txt",
+		"/buckets/test-bucket/d01/s00/f000.txt",
+	}, collect("d00/s01"))
+
+	assert.Equal(t, []string{
+		"/buckets/test-bucket/d01/s00/f000.txt",
+	}, collect("d01"))
+}
+
 func TestDoListFilerEntries_EmptyDirectoriesDoNotHideLaterEntries(t *testing.T) {
 	// With maxKeys=1 the listing window is 3 entries. The first window holds only
 	// empty directories, so the real object further down must still be found.