s3: list keys in byte order around directory markers (#11832)

* s3: list keys in byte order around directory markers

* s3: order deferred dir blocks by slash name and hold quota

* s3: fix deferred dir blocks at page and marker boundaries

Review follow-ups on the marker byte-order change:

- A marker inside an emitted "<name>/" block re-descended and re-listed its
  folded contents under delimiter "/" instead of skipping the block.
- A prefix object's "<name>/" common prefix was emitted eagerly at the name
  position, outranking siblings like "a.txt" and repeating across pages; the
  plain key now emits alone at the name position and the block emits the
  common prefix when children exist.
- A recursion starved on quota before listing a deferred subtree returned
  without truncation, dropping keys silently; the block now reports a
  resumable truncation when the subtree still holds entries.
- Clone prefix-object entries with proto.Clone rather than copying the
  protobuf message by value.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* s3: close the hasEntries probe stream on return

* s3: resume delimited listings at the dir block boundary

A deferred directory block emits at its "<name>/" position, but emitDir
left nextMarker at the bare name, which sorts inside the (name, name/)
band and reads as "block pending". Every pass of the listing loop then
re-emitted the same common prefix until maxKeys ran out (999 prefixes,
IsTruncated), and a "<name>/" continuation token trimmed to "<name>" by
adjustMarkerForDelimiter reopened the block and paged forever.

Advance nextMarker to the block boundary the same way the recursive
branch does, and keep a "/" -ending marker intact — it already names the
bound past the block.

---------

Co-authored-by: Devin <devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/s3api/s3api_object_handlers_list.go

@@ -19,6 +19,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 )
 
 type OptionalString struct {
@@ -565,6 +566,8 @@ func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, req listObjectsReq
 					// plain key. The traversal never descends into it under a delimiter,
 					// so the CommonPrefix its nested keys fold into is added here too.
 					if entry.IsPrefixObject() {
+						keyOnly := cursor.prefixObjectKeyOnly
+						cursor.prefixObjectKeyOnly = false
 						key := fmt.Sprintf("%s/%s", dir, entry.Name)[len(bucketPrefix):]
 						if strings.HasPrefix(key, originalPrefix) {
 							if folded := prefixForKey(dir, entry.Name); folded != "" {
@@ -579,8 +582,10 @@ func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, req listObjectsReq
 						// whole, so a page ending between them cannot resume at the prefix:
 						// a marker of "<key>/" means the subtree is done, not that it is
 						// next. The page runs one item over maxKeys instead of dropping it.
-						if childPrefix := prefixForKey(dir, entry.Name+"/"); childPrefix != "" && s3a.hasChildren(ctx, bucket, key) {
-							addCommonPrefix(childPrefix)
+						if !keyOnly {
+							if childPrefix := prefixForKey(dir, entry.Name+"/"); childPrefix != "" && s3a.hasChildren(ctx, bucket, key) {
+								addCommonPrefix(childPrefix)
+							}
 						}
 						return
 					}
@@ -804,6 +809,10 @@ type ListingCursor struct {
 	// emitDirAsCommonPrefix marks the next directory entry passed to the entry
 	// callback for reporting as a CommonPrefix rather than a Contents key.
 	emitDirAsCommonPrefix bool
+	// prefixObjectKeyOnly limits the next prefix-object entry passed to the entry
+	// callback to its plain key; its "<name>/" CommonPrefix is deferred to the
+	// directory's own block position.
+	prefixObjectKeyOnly bool
 }
 
 // excludedMarkerKey returns the key an exclusive marker names that the walk's cutoff no
@@ -950,35 +959,20 @@ 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)
-		subNextMarker, subErr := s3a.doListFilerEntries(ctx, client, listDirectoryRequest{dir: dir + "/" + subDir, marker: subMarker, delimiter: delimiter, bucket: bucket}, cursor, eachEntryFn)
-		if subErr != nil {
-			err = subErr
-			return
+	// A directory's keys sort at "<name>/" among this level's children, not at
+	// the name itself: siblings like "a-c" and "a.txt" sort before "a/b" while
+	// "a0" sorts after. Start the window at the first child whose block can
+	// still hold keys past the marker — the shortest prefix of the marker
+	// followed by a byte no greater than "/" — and ask for it inclusively, so
+	// a marker-named directory's subtree is reached at all.
+	startName := marker
+	for i := 0; i < len(marker); i++ {
+		if marker[i] <= '/' {
+			startName = marker[:i]
+			break
 		}
-		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
 	}
+	inclusiveStartFrom = startName != ""
 
 	// now marker is also a direct child of dir
 	ctx, cancel := context.WithCancel(ctx)
@@ -988,6 +982,152 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.
 	// each request window inside the page.
 	pageMarker := marker
 
+	// Defer each directory until a child past its "<name>/" block arrives, so
+	// emission follows S3 byte order rather than the filer's name order.
+	type pendingDir struct {
+		e                *filer_pb.Entry
+		matchedPrefixDir bool
+	}
+	var pendingDirs []pendingDir
+	stopped := false
+	// hasEntries reports whether the filer directory holds at least one entry,
+	// probed on the request's own client; errors answer conservatively as true.
+	hasEntries := func(path string) bool {
+		probeCtx, cancel := context.WithCancel(ctx)
+		defer cancel()
+		stream, lerr := client.ListEntries(probeCtx, &filer_pb.ListEntriesRequest{Directory: path, Limit: 1})
+		if lerr != nil {
+			return true
+		}
+		_, recvErr := stream.Recv()
+		return recvErr == nil
+	}
+	emitDir := func(e *filer_pb.Entry, matchedPrefixDir bool) error {
+		blockKey := e.Name + "/"
+		var subMarker string
+		if strings.HasPrefix(marker, blockKey) {
+			if delimiter == "/" {
+				// this block collapses to one common prefix, already emitted
+				return nil
+			}
+			subMarker = marker[len(blockKey):]
+		} else if blockKey <= marker {
+			return nil
+		}
+		// A prefix object's plain key is emitted at the name position or not at
+		// all; the block emits only its "<name>/" fold, so the callback sees a
+		// plain directory.
+		emitEntry := e
+		if e.IsPrefixObject() {
+			emitEntry = proto.Clone(e).(*filer_pb.Entry)
+			delete(emitEntry.Extended, s3_constants.SeaweedFSPrefixObject)
+		}
+		isKeyObject := e.IsDirectoryKeyObject() && !e.IsPrefixObject()
+		if delimiter != "/" || cursor.prefixEndsOnDelimiter {
+			// A trailing-slash prefix (e.g. "logs/") names one directory and asks
+			// whether it exists, so a real but empty directory must be reported for
+			// that probe.
+			explicitDirProbe := cursor.prefixEndsOnDelimiter
+			if cursor.prefixEndsOnDelimiter {
+				cursor.prefixEndsOnDelimiter = false
+			}
+			if isKeyObject && blockKey > marker {
+				// Directory key objects (created via PutObject with trailing "/")
+				// appear as regular keys in recursive listing mode.
+				eachEntryFn(dir, emitEntry)
+			}
+			// Recurse into subdirectory to list any children, noting whether the
+			// subtree produced any entries.
+			childEmitted := false
+			subNextMarker, subErr := s3a.doListFilerEntries(ctx, client, listDirectoryRequest{dir: dir + "/" + e.Name, marker: subMarker, delimiter: delimiter, bucket: bucket}, cursor, func(d string, ent *filer_pb.Entry) {
+				childEmitted = true
+				eachEntryFn(d, ent)
+			})
+			if subErr != nil {
+				return fmt.Errorf("doListFilerEntries2: %w", subErr)
+			}
+			nextMarker = e.Name + "/" + subNextMarker
+			if cursor.maxKeys <= 0 && subNextMarker == "" && hasEntries(dir+"/"+e.Name) {
+				// The recursion starved on quota before listing the subtree;
+				// keep a resume position that re-enters it.
+				cursor.isTruncated = true
+			}
+			// A real but empty directory (created out of band via mount, mkdir or
+			// the filer API, so it carries no MIME) is otherwise invisible to S3
+			// clients that detect directories by listing it under its own "<dir>/"
+			// prefix. Answer that explicit probe so tools like hadoop-aws can find
+			// the directory: a delimited listing reports it as a CommonPrefix --
+			// a real key in Contents would look like an object a client can read --
+			// while a recursive listing keeps the directory marker key it used.
+			// Plain listings are left untouched, so empty directories left behind
+			// by deleted objects are not shown as keys.
+			// A directory that still holds version history no longer names anything,
+			// so it gets no marker either.
+			if explicitDirProbe && !isKeyObject && !childEmitted && !cursor.isTruncated && e.Attributes != nil &&
+				!s3a.dirHoldsOnlyHiddenEntries(ctx, client, bucket, dir+"/"+e.Name, cursor) {
+				if delimiter == "/" {
+					cursor.emitDirAsCommonPrefix = true
+				} else {
+					emitEntry.Attributes.Mime = s3_constants.FolderMimeType
+				}
+				eachEntryFn(dir, emitEntry)
+			}
+			// println("doListFilerEntries2 dir", dir+"/"+e.Name, "subNextMarker", subNextMarker)
+			if cursor.isTruncated || matchedPrefixDir {
+				stopped = true
+			}
+			// println("doListFilerEntries2 nextMarker", nextMarker)
+		} else {
+			switch {
+			case isKeyObject && blockKey > marker:
+				eachEntryFn(dir, emitEntry)
+			case e.IsPrefixObject():
+				// The plain key already listed at the name position; the
+				// deferred block emits the "<name>/" common prefix only when
+				// children fold under it.
+				if hasEntries(dir + "/" + e.Name) {
+					eachEntryFn(dir, emitEntry)
+				}
+			case !s3a.dirHoldsOnlyHiddenEntries(ctx, client, bucket, dir+"/"+e.Name, cursor):
+				eachEntryFn(dir, emitEntry)
+			}
+			// Whatever the block emitted sits at its "<name>/" position, so the
+			// resume marker is the block boundary: the bare name sorts inside the
+			// (name, name/) band and would reopen the block for a second emission.
+			nextMarker = e.Name + "/"
+		}
+		return nil
+	}
+	// pendingDirs is kept ordered by "<name>/" rather than by name: "a-c/" and
+	// "a./" sort before "a/" even though "a" is the smaller name.
+	flushDirs := func(upTo string) error {
+		for len(pendingDirs) > 0 && (upTo == "" || pendingDirs[0].e.Name+"/" < upTo) {
+			if cursor.maxKeys <= 0 {
+				cursor.isTruncated = true
+				return nil
+			}
+			d := pendingDirs[0]
+			pendingDirs = pendingDirs[1:]
+			if err := emitDir(d.e, d.matchedPrefixDir); err != nil {
+				return err
+			}
+			if stopped || cursor.isTruncated {
+				return nil
+			}
+		}
+		return nil
+	}
+	queueDir := func(pd pendingDir) {
+		blockKey := pd.e.Name + "/"
+		i := 0
+		for i < len(pendingDirs) && pendingDirs[i].e.Name+"/" < blockKey {
+			i++
+		}
+		pendingDirs = append(pendingDirs, pendingDir{})
+		copy(pendingDirs[i+1:], pendingDirs[i:])
+		pendingDirs[i] = pd
+	}
+
 	// Entries that emit nothing (empty directories, the .uploads folder, the marker
 	// echo) consume the request window without consuming maxKeys, so one window may
 	// end before maxKeys is satisfied. Keep requesting from the last received entry
@@ -997,7 +1137,7 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.
 			Directory:          dir,
 			Prefix:             prefix,
 			Limit:              uint32(cursor.maxKeys) + 2, // bucket root directory needs to skip additional s3_constants.MultipartUploadsFolder folder
-			StartFromFileName:  marker,
+			StartFromFileName:  startName,
 			InclusiveStartFrom: inclusiveStartFrom,
 		}
 
@@ -1028,47 +1168,21 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.
 			}
 			entriesReceived++
 			lastEntryName = entry.Name
-			// listFilerEntries always calls doListFilerEntries with inclusiveStartFrom=false
-			// (S3 marker semantics are exclusive), but keep the guard explicit to preserve
-			// behavior if inclusive callers are introduced in the future.
-			// A versioned object lives in a "<key>.versions" directory, so the marker also
-			// has to be matched against the object name that directory stands for.
-			markerName := entry.Name
-			if entry.IsDirectory {
-				markerName = strings.TrimSuffix(markerName, s3_constants.VersionsFolder)
+			// A directory whose block sorts before this entry emits now — every
+			// key it yields still sorts before this entry's name.
+			if err = flushDirs(entry.Name); err != nil || stopped || cursor.isTruncated {
+				return
 			}
-			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 {
+			// A directory block still contributes past a marker that lands inside
+			// its "<name>/" band; only a block entirely behind it is skipped.
+			if marker != "" {
+				if entry.IsDirectory {
+					if entry.Name+"/" <= marker && !strings.HasPrefix(marker, entry.Name+"/") {
+						continue
+					}
+				} else if entry.Name <= marker {
 					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
 			}
 
 			// The .versions sibling of the key just emitted still decides that key's
@@ -1105,6 +1219,25 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.
 			}
 			if entry.IsDirectory {
 				// glog.V(4).Infof("List Dir Entries %s, file: %s, maxKeys %d", dir, entry.Name, cursor.maxKeys)
+				// A marker inside a skipped subtree still descends — the
+				// internal folder is not listed, its contents may be.
+				if strings.HasPrefix(marker, entry.Name+"/") {
+					if delimiter == "/" {
+						// The whole block already collapsed into its common
+						// prefix; nothing under it is listed again.
+						continue
+					}
+					subNextMarker, subErr := s3a.doListFilerEntries(ctx, client, listDirectoryRequest{dir: dir + "/" + entry.Name, marker: marker[len(entry.Name)+1:], 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
+				}
 				if entry.Name == s3_constants.MultipartUploadsFolder { // FIXME no need to apply to all directories. this extra also affects maxKeys
 					continue
 				}
@@ -1119,6 +1252,10 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.
 					}
 					// Extract object name from .versions directory name
 					baseObjectName := strings.TrimSuffix(entry.Name, s3_constants.VersionsFolder)
+					// A marker at or past the folded key already covered it.
+					if marker != "" && baseObjectName <= marker {
+						continue
+					}
 					// A page resuming from a marker inside the base key's extension
 					// region ("k.bak" sorts between "k" and "k.versions") means an
 					// earlier page already listed and settled the base null object;
@@ -1155,76 +1292,32 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.
 					continue
 				}
 
-				if delimiter != "/" || cursor.prefixEndsOnDelimiter {
-					// A trailing-slash prefix (e.g. "logs/") names one directory and asks
-					// whether it exists, so a real but empty directory must be reported for
-					// that probe.
-					explicitDirProbe := cursor.prefixEndsOnDelimiter
-					if cursor.prefixEndsOnDelimiter {
-						cursor.prefixEndsOnDelimiter = false
-					}
-					isKeyObject := entry.IsDirectoryKeyObject()
-					if isKeyObject {
-						// Directory key objects (created via PutObject with trailing "/")
-						// must appear as regular keys in recursive listing mode.
-						eachEntryFn(dir, entry)
-					}
-					// Recurse into subdirectory to list any children, noting whether the
-					// subtree produced any entries.
-					childEmitted := false
-					subNextMarker, subErr := s3a.doListFilerEntries(ctx, client, listDirectoryRequest{dir: dir + "/" + entry.Name, delimiter: delimiter, bucket: bucket}, cursor, func(d string, e *filer_pb.Entry) {
-						childEmitted = true
-						eachEntryFn(d, e)
-					})
-					if subErr != nil {
-						err = fmt.Errorf("doListFilerEntries2: %w", subErr)
-						return
-					}
-					// A real but empty directory (created out of band via mount, mkdir or
-					// the filer API, so it carries no MIME) is otherwise invisible to S3
-					// clients that detect directories by listing it under its own "<dir>/"
-					// prefix. Answer that explicit probe so tools like hadoop-aws can find
-					// the directory: a delimited listing reports it as a CommonPrefix --
-					// a real key in Contents would look like an object a client can read --
-					// while a recursive listing keeps the directory marker key it used.
-					// Plain listings are left untouched, so empty directories left behind
-					// by deleted objects are not shown as keys.
-					// A directory that still holds version history no longer names anything,
-					// so it gets no marker either.
-					if explicitDirProbe && !isKeyObject && !childEmitted && !cursor.isTruncated && entry.Attributes != nil &&
-						!s3a.dirHoldsOnlyHiddenEntries(ctx, client, bucket, dir+"/"+entry.Name, cursor) {
-						if delimiter == "/" {
-							cursor.emitDirAsCommonPrefix = true
-						} else {
-							entry.Attributes.Mime = s3_constants.FolderMimeType
-						}
-						eachEntryFn(dir, entry)
-					}
-					// println("doListFilerEntries2 dir", dir+"/"+entry.Name, "subNextMarker", subNextMarker)
-					nextMarker = entry.Name + "/" + subNextMarker
-					if cursor.isTruncated {
-						return
-					}
-					if matchedPrefixDir {
-						return
-					}
-					// println("doListFilerEntries2 nextMarker", nextMarker)
-				} else if entry.IsDirectoryKeyObject() || !s3a.dirHoldsOnlyHiddenEntries(ctx, client, bucket, dir+"/"+entry.Name, cursor) {
+				pd := pendingDir{e: entry, matchedPrefixDir: matchedPrefixDir}
+				// A prefix object lists as the plain key "<name>", which sorts
+				// at the name, not at the "<name>/" block its children occupy; the
+				// deferred block emits the common prefix at its own position.
+				if entry.IsPrefixObject() && entry.Name > marker {
+					cursor.prefixObjectKeyOnly = true
 					eachEntryFn(dir, entry)
 				}
-			} else {
-				eachEntryFn(dir, entry)
-				// glog.V(4).Infof("List File Entries %s, file: %s, maxKeys %d", dir, entry.Name, cursor.maxKeys)
+				queueDir(pd)
+				continue
 			}
+			eachEntryFn(dir, entry)
+			// glog.V(4).Infof("List File Entries %s, file: %s, maxKeys %d", dir, entry.Name, cursor.maxKeys)
 			if cursor.prefixEndsOnDelimiter {
 				cursor.prefixEndsOnDelimiter = false
 			}
 		}
 
-		if cursor.isTruncated || entriesReceived < request.Limit {
+		if cursor.isTruncated || stopped {
 			return
 		}
-		marker = lastEntryName
+		if entriesReceived < request.Limit {
+			err = flushDirs("")
+			return
+		}
+		startName = lastEntryName
 		inclusiveStartFrom = false
 	}
 }
@@ -1482,6 +1575,15 @@ func adjustMarkerForDelimiter(marker, prefix, delimiter string) string {
 		return marker
 	}
 
+	// A marker ending on "/" already names the bound past that directory's
+	// block — the walk reads "boo/" as "the boo/ block is done". Trimming it
+	// to the bare name would sort inside the (name, name/) band instead and
+	// reopen the block. Other delimiters have no matching directory boundary,
+	// so their markers still need the trim below.
+	if delimiter == "/" {
+		return marker
+	}
+
 	// Remove the trailing delimiter
 	// This ensures we skip all entries under the prefix but don't skip
 	// potential directory entries that start with a similar prefix

weed/s3api/s3api_object_handlers_list_test.go

@@ -4,11 +4,13 @@ import (
 	"context"
 	"io"
 	"net/http/httptest"
+	"os"
 	"strings"
 	"testing"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
 	"github.com/stretchr/testify/assert"
 	grpc "google.golang.org/grpc"
@@ -98,18 +100,31 @@ func (c *markerEchoFilerClient) ListEntries(ctx context.Context, in *filer_pb.Li
 		return &testListEntriesStream{entries: entries}, nil
 	}
 
-	for i, e := range entries {
-		if e.Name == in.StartFromFileName {
-			// Emulate buggy backend behavior: return marker again even when exclusive.
-			if c.returnFollowing {
-				echoAndFollowing := entries[i:]
-				return &testListEntriesStream{entries: echoAndFollowing}, nil
-			}
-			return &testListEntriesStream{entries: []*filer_pb.Entry{e}}, nil
+	filtered := make([]*filer_pb.Entry, 0, len(entries))
+	for _, e := range entries {
+		if e.Name > in.StartFromFileName || (in.InclusiveStartFrom && e.Name == in.StartFromFileName) {
+			filtered = append(filtered, e)
 		}
 	}
-
-	return &testListEntriesStream{entries: nil}, nil
+	// Emulate buggy backend behavior: echo the start entry again even when the
+	// request did not ask for it.
+	if !in.InclusiveStartFrom {
+		var echo *filer_pb.Entry
+		for _, e := range entries {
+			if e.Name == in.StartFromFileName {
+				echo = e
+				break
+			}
+		}
+		if echo != nil {
+			if c.returnFollowing {
+				filtered = append([]*filer_pb.Entry{echo}, filtered...)
+			} else {
+				filtered = []*filer_pb.Entry{echo}
+			}
+		}
+	}
+	return &testListEntriesStream{entries: filtered}, nil
 }
 
 func ensureEntryAttributes(entries []*filer_pb.Entry) {
@@ -1143,3 +1158,323 @@ func TestListObjectsV2_PrefixEndingWithSlash_WithDelimiter(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, []string{"fileA"}, results, "Should only return files under directory '1', not '1000'")
 }
+
+// A marker landing between a directory's name and its trailing slash must
+// still produce keys inside the directory in byte order: the marker's own
+// block band, not just its name, decides what follows it.
+func TestDoListFilerEntries_MarkerBetweenDirAndSlash(t *testing.T) {
+	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/b/o":     {dir("a"), file("a-c"), file("a.txt"), file("a0"), file("ab")},
+			"/buckets/b/o/a":   {file("b"), dir("c")},
+			"/buckets/b/o/a/c": {file("d")},
+		},
+	}
+
+	collect := func(marker string) []string {
+		cursor := &ListingCursor{maxKeys: 1000}
+		var seen []string
+		_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/b/o", marker: marker, bucket: "b"}, cursor, func(dir string, entry *filer_pb.Entry) {
+			seen = append(seen, strings.TrimPrefix(dir, "/buckets/b/")+"/"+entry.Name)
+			cursor.maxKeys--
+		})
+		assert.NoError(t, err)
+		return seen
+	}
+
+	// byte order: o/a-c, o/a.txt, o/a/b, o/a/c/d, o/a0, o/ab
+	assert.Equal(t, []string{
+		"o/a-c", "o/a.txt", "o/a/b", "o/a/c/d", "o/a0", "o/ab",
+	}, collect(""))
+
+	assert.Equal(t, []string{
+		"o/a/b", "o/a/c/d", "o/a0", "o/ab",
+	}, collect("a.txt"))
+
+	assert.Equal(t, []string{
+		"o/a/c/d", "o/a0", "o/ab",
+	}, collect("a/b"))
+
+	assert.Equal(t, []string{
+		"o/ab",
+	}, collect("a0"))
+
+	// a marker that sorts inside the "a" band before its slash still walks a:
+	assert.Equal(t, []string{
+		"o/a.txt", "o/a/b", "o/a/c/d", "o/a0", "o/ab",
+	}, collect("a-c"))
+}
+
+// Directory blocks order by "<name>/", not by name: "a-/" sorts before "a/".
+func TestDoListFilerEntries_DirBlocksSortBySlashName(t *testing.T) {
+	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/b":    {dir("a"), dir("a-"), file("a.txt")},
+			"/buckets/b/a":  {file("b")},
+			"/buckets/b/a-": {file("x")},
+		},
+	}
+
+	cursor := &ListingCursor{maxKeys: 1000}
+	var seen []string
+	_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/b", bucket: "b"}, cursor, func(dir string, entry *filer_pb.Entry) {
+		seen = append(seen, strings.TrimPrefix(dir, "/buckets/b")+"/"+entry.Name)
+		cursor.maxKeys--
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, []string{"/a-/x", "/a.txt", "/a/b"}, seen)
+}
+
+// A full page must report truncation while deferred directories still hold keys.
+func TestDoListFilerEntries_FullPageTruncatesWithPendingDir(t *testing.T) {
+	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/b":   {dir("a"), file("a-c")},
+			"/buckets/b/a": {file("x")},
+		},
+	}
+
+	cursor := &ListingCursor{maxKeys: 1}
+	var seen []string
+	nextMarker, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/b", bucket: "b"}, cursor, func(dir string, entry *filer_pb.Entry) {
+		seen = append(seen, entry.Name)
+		cursor.maxKeys--
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, []string{"a-c"}, seen)
+	assert.True(t, cursor.isTruncated, "deferred directory still holds a key")
+	assert.Equal(t, "a-c", nextMarker)
+
+	// the next page re-enters the deferred block through the marker band
+	cursor = &ListingCursor{maxKeys: 1000}
+	seen = nil
+	_, err = s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/b", marker: nextMarker, bucket: "b"}, cursor, func(dir string, entry *filer_pb.Entry) {
+		seen = append(seen, dir+"/"+entry.Name)
+		cursor.maxKeys--
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, []string{"/buckets/b/a/x"}, seen)
+}
+
+// A directory that is also a plain object lists "a" at the name position and
+// does not repeat it when the marker is the object itself.
+func TestDoListFilerEntries_PrefixObjectKeepsNamePosition(t *testing.T) {
+	s3a := &S3ApiServer{}
+	dirObj := func(name string) *filer_pb.Entry {
+		e := &filer_pb.Entry{Name: name, IsDirectory: true, Attributes: &filer_pb.FuseAttributes{FileMode: uint32(os.ModeDir | 0644)}}
+		e.MarkPrefixObject()
+		return e
+	}
+	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/b":   {dirObj("a"), file("a.txt")},
+			"/buckets/b/a": {file("b")},
+		},
+	}
+
+	collect := func(marker string) []string {
+		cursor := &ListingCursor{maxKeys: 1000}
+		var seen []string
+		_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/b", marker: marker, bucket: "b"}, 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/b/a", "/buckets/b/a.txt", "/buckets/b/a/b"}, collect(""))
+	// the marker is the object "a" itself: excluded, but "a.txt" and the
+	// "a/" block still follow
+	assert.Equal(t, []string{"/buckets/b/a.txt", "/buckets/b/a/b"}, collect("a"))
+}
+
+// A marker that lands inside a directory's "<name>/" block — the continuation
+// token a delimited page returns for it — must not re-list the block's folded
+// contents under delimiter "/".
+func TestDoListFilerEntries_DelimiterSkipsResumedDirBlock(t *testing.T) {
+	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/b":   {dir("a"), file("b")},
+			"/buckets/b/a": {file("x")},
+		},
+	}
+
+	collect := func(marker string) []string {
+		cursor := &ListingCursor{maxKeys: 1000}
+		var seen []string
+		_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/b", marker: marker, delimiter: "/", bucket: "b"}, cursor, func(dir string, entry *filer_pb.Entry) {
+			seen = append(seen, strings.TrimPrefix(dir, "/buckets/b")+"/"+entry.Name)
+			cursor.maxKeys--
+		})
+		assert.NoError(t, err)
+		return seen
+	}
+
+	// The "a/" block already collapsed into a common prefix on the earlier
+	// page, so neither "a/" nor anything under it is emitted again.
+	assert.Equal(t, []string{"/b"}, collect("a/"))
+	assert.Equal(t, []string{"/b"}, collect("a/x"))
+}
+
+// A delimited listing resumes a directory at its "<name>/" block boundary, not
+// the bare name: the name alone sorts inside the (name, name/) band and makes
+// the next pass re-emit the same common prefix until the quota runs out. The
+// marker also survives a round trip through the request path unchanged —
+// adjustMarkerForDelimiter must not trim the trailing slash off it.
+func TestDoListFilerEntries_DelimitedDirResumesAtBlockBoundary(t *testing.T) {
+	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/b":   {dir("a"), file("b")},
+			"/buckets/b/a": {file("x")},
+		},
+	}
+
+	// The outer listFilerEntries loop re-enters with the returned marker until
+	// a pass comes back empty; every entry must be emitted exactly once.
+	var seen []string
+	marker := ""
+	for pass := 0; pass < 4; pass++ {
+		cursor := &ListingCursor{maxKeys: 1000}
+		emitted := 0
+		nextMarker, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/b", marker: marker, delimiter: "/", bucket: "b"}, cursor, func(dir string, entry *filer_pb.Entry) {
+			seen = append(seen, strings.TrimPrefix(dir, "/buckets/b")+"/"+entry.Name)
+			emitted++
+		})
+		assert.NoError(t, err)
+		if emitted == 0 {
+			break
+		}
+		marker = adjustMarkerForDelimiter(nextMarker, "", "/")
+	}
+	assert.Equal(t, []string{"/a", "/b"}, seen)
+}
+
+// A prefix object's "<name>/" common prefix belongs to the block position, not
+// the name position: it must not outrank "a.txt" on a full page nor repeat on
+// the next one.
+func TestDoListFilerEntries_PrefixObjectDefersCommonPrefix(t *testing.T) {
+	s3a := &S3ApiServer{}
+	dirObj := func(name string) *filer_pb.Entry {
+		e := &filer_pb.Entry{Name: name, IsDirectory: true, Attributes: &filer_pb.FuseAttributes{FileMode: uint32(os.ModeDir | 0644)}}
+		e.MarkPrefixObject()
+		return e
+	}
+	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/b":   {dirObj("a"), file("a.txt")},
+			"/buckets/b/a": {file("b")},
+		},
+	}
+
+	// Page one holds the plain key "a" and "a.txt"; the "a/" block sort after
+	// both, so the page truncates at "a.txt" without touching it.
+	cursor := &ListingCursor{maxKeys: 2}
+	var seen []string
+	var sawPrefixObject bool
+	nextMarker, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/b", delimiter: "/", bucket: "b"}, cursor, func(dir string, entry *filer_pb.Entry) {
+		seen = append(seen, entry.Name)
+		sawPrefixObject = sawPrefixObject || entry.IsPrefixObject()
+		cursor.maxKeys--
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, []string{"a", "a.txt"}, seen)
+	assert.True(t, sawPrefixObject, "the name-position emission is the plain object")
+	assert.True(t, cursor.isTruncated)
+	assert.Equal(t, "a.txt", nextMarker)
+
+	// Page two emits the deferred block once, as a plain directory fold.
+	cursor = &ListingCursor{maxKeys: 2}
+	seen = nil
+	sawPrefixObject = false
+	_, err = s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/b", marker: nextMarker, delimiter: "/", bucket: "b"}, cursor, func(dir string, entry *filer_pb.Entry) {
+		seen = append(seen, entry.Name)
+		sawPrefixObject = sawPrefixObject || entry.IsPrefixObject()
+		cursor.maxKeys--
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, []string{"a"}, seen)
+	assert.False(t, sawPrefixObject, "the block emission is the \"<name>/\" fold, not the plain key again")
+	assert.False(t, cursor.isTruncated)
+}
+
+// A directory key object that spends the page's last slot still reports
+// truncation so the next page re-enters its subtree.
+func TestDoListFilerEntries_DirKeyObjectBoundaryTruncates(t *testing.T) {
+	s3a := &S3ApiServer{}
+	dirKeyObj := func(name string) *filer_pb.Entry {
+		return &filer_pb.Entry{Name: name, IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mime: s3_constants.FolderMimeType}}
+	}
+	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/b":   {dirKeyObj("a")},
+			"/buckets/b/a": {file("x")},
+		},
+	}
+
+	cursor := &ListingCursor{maxKeys: 1}
+	var seen []string
+	nextMarker, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/b", bucket: "b"}, cursor, func(dir string, entry *filer_pb.Entry) {
+		seen = append(seen, entry.Name)
+		cursor.maxKeys--
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, []string{"a"}, seen)
+	assert.True(t, cursor.isTruncated, "the subtree may still hold unlisted keys")
+	assert.Equal(t, "a/", nextMarker)
+
+	// The next page resumes inside the "a/" block and lists the child.
+	cursor = &ListingCursor{maxKeys: 1000}
+	seen = nil
+	_, err = s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/b", marker: nextMarker, bucket: "b"}, cursor, func(dir string, entry *filer_pb.Entry) {
+		seen = append(seen, strings.TrimPrefix(dir, "/buckets/b")+"/"+entry.Name)
+		cursor.maxKeys--
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, []string{"/a/x"}, seen)
+	assert.False(t, cursor.isTruncated)
+}