diff --git a/weed/s3api/s3api_object_handlers_list.go b/weed/s3api/s3api_object_handlers_list.go
index d77eb3d32..9fe39618b 100644
--- a/weed/s3api/s3api_object_handlers_list.go
+++ b/weed/s3api/s3api_object_handlers_list.go
@@ -109,7 +109,15 @@ func (s3a *S3ApiServer) ListObjectsV2Handler(w http.ResponseWriter, r *http.Requ
 	// Adjust marker if it ends with delimiter to skip all entries with that prefix
 	marker = adjustMarkerForDelimiter(marker, delimiter)
 
-	response, err := s3a.listFilerEntries(r.Context(), bucket, originalPrefix, maxKeys, marker, delimiter, encodingTypeUrl, fetchOwner)
+	response, err := s3a.listFilerEntries(r.Context(), listObjectsRequest{
+		bucket:          bucket,
+		prefix:          originalPrefix,
+		marker:          marker,
+		delimiter:       delimiter,
+		maxKeys:         maxKeys,
+		encodingTypeUrl: encodingTypeUrl,
+		fetchOwner:      fetchOwner,
+	})
 
 	if err != nil {
 		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
@@ -173,7 +181,15 @@ func (s3a *S3ApiServer) ListObjectsV1Handler(w http.ResponseWriter, r *http.Requ
 	// Adjust marker if it ends with delimiter to skip all entries with that prefix
 	marker = adjustMarkerForDelimiter(marker, delimiter)
 
-	response, err := s3a.listFilerEntries(r.Context(), bucket, originalPrefix, uint16(maxKeys), marker, delimiter, encodingTypeUrl, true)
+	response, err := s3a.listFilerEntries(r.Context(), listObjectsRequest{
+		bucket:          bucket,
+		prefix:          originalPrefix,
+		marker:          marker,
+		delimiter:       delimiter,
+		maxKeys:         uint16(maxKeys),
+		encodingTypeUrl: encodingTypeUrl,
+		fetchOwner:      true,
+	})
 
 	if err != nil {
 		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
@@ -232,7 +248,20 @@ func sanitizeV1MarkerEcho(response *ListBucketResult, marker string, encodingTyp
 	}
 }
 
-func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, bucket string, originalPrefix string, maxKeys uint16, originalMarker string, delimiter string, encodingTypeUrl bool, fetchOwner bool) (response ListBucketResult, err error) {
+type listObjectsRequest struct {
+	bucket          string
+	prefix          string
+	marker          string
+	delimiter       string
+	maxKeys         uint16
+	encodingTypeUrl bool
+	fetchOwner      bool
+}
+
+func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, req listObjectsRequest) (response ListBucketResult, err error) {
+	bucket, originalPrefix, originalMarker := req.bucket, req.prefix, req.marker
+	maxKeys, delimiter := req.maxKeys, req.delimiter
+	encodingTypeUrl, fetchOwner := req.encodingTypeUrl, req.fetchOwner
 	// convert full path prefix into directory name and prefix for entry name
 	requestDir, prefix, marker := normalizePrefixMarker(originalPrefix, originalMarker)
 	bucketPrefix := s3a.bucketPrefix(bucket)
@@ -298,7 +327,7 @@ func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, bucket string, ori
 		for {
 			empty := true
 
-			nextMarker, doErr = s3a.doListFilerEntries(client, reqDir, prefix, cursor, marker, delimiter, false, bucket, func(dir string, entry *filer_pb.Entry) {
+			nextMarker, doErr = s3a.doListFilerEntries(ctx, client, listDirectoryRequest{dir: reqDir, prefix: prefix, marker: marker, delimiter: delimiter, bucket: bucket}, cursor, func(dir string, entry *filer_pb.Entry) {
 				empty = false
 				dirName, entryName, _ := entryUrlEncode(dir, entry.Name, encodingTypeUrl)
 				if entry.IsDirectory {
@@ -557,7 +586,18 @@ func buildTruncatedNextMarker(requestDir, prefix, nextMarker string, lastEntryWa
 	return nextMarker
 }
 
-func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, dir, prefix string, cursor *ListingCursor, marker, delimiter string, inclusiveStartFrom bool, bucket string, eachEntryFn func(dir string, entry *filer_pb.Entry)) (nextMarker string, err error) {
+type listDirectoryRequest struct {
+	dir                string
+	prefix             string
+	marker             string
+	delimiter          string
+	bucket             string
+	inclusiveStartFrom bool
+}
+
+func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.SeaweedFilerClient, req listDirectoryRequest, cursor *ListingCursor, eachEntryFn func(dir string, entry *filer_pb.Entry)) (nextMarker string, err error) {
+	dir, prefix, bucket := req.dir, req.prefix, req.bucket
+	marker, delimiter, inclusiveStartFrom := req.marker, req.delimiter, req.inclusiveStartFrom
 	// invariants
 	//   prefix and marker should be under dir, marker may contain "/"
 	//   maxKeys should be updated for each recursion
@@ -571,7 +611,7 @@ func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, d
 	if strings.Contains(marker, "/") {
 		subDir, subMarker := toParentAndDescendants(marker)
 		// println("doListFilerEntries dir", dir+"/"+subDir, "subMarker", subMarker)
-		subNextMarker, subErr := s3a.doListFilerEntries(client, dir+"/"+subDir, "", cursor, subMarker, delimiter, false, bucket, eachEntryFn)
+		subNextMarker, subErr := s3a.doListFilerEntries(ctx, client, listDirectoryRequest{dir: dir + "/" + subDir, marker: subMarker, delimiter: delimiter, bucket: bucket}, cursor, eachEntryFn)
 		if subErr != nil {
 			err = subErr
 			return
@@ -585,156 +625,170 @@ func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, d
 	}
 
 	// now marker is also a direct child of dir
-	request := &filer_pb.ListEntriesRequest{
-		Directory:          dir,
-		Prefix:             prefix,
-		Limit:              uint32(cursor.maxKeys) + 2, // bucket root directory needs to skip additional s3_constants.MultipartUploadsFolder folder
-		StartFromFileName:  marker,
-		InclusiveStartFrom: inclusiveStartFrom,
-	}
-
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
-	stream, listErr := client.ListEntries(ctx, request)
-	if listErr != nil {
-		if errors.Is(listErr, filer_pb.ErrNotFound) {
-			return
-		}
-		err = fmt.Errorf("list entries %+v: %w", request, listErr)
-		return
-	}
 
+	// 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
+	// until the quota is filled or a short window shows the directory is exhausted.
 	for {
-		resp, recvErr := stream.Recv()
-		if recvErr != nil {
-			if recvErr == io.EOF {
-				break
-			} else {
-				err = fmt.Errorf("iterating entries %+v: %v", request, recvErr)
+		request := &filer_pb.ListEntriesRequest{
+			Directory:          dir,
+			Prefix:             prefix,
+			Limit:              uint32(cursor.maxKeys) + 2, // bucket root directory needs to skip additional s3_constants.MultipartUploadsFolder folder
+			StartFromFileName:  marker,
+			InclusiveStartFrom: inclusiveStartFrom,
+		}
+
+		stream, listErr := client.ListEntries(ctx, request)
+		if listErr != nil {
+			if errors.Is(listErr, filer_pb.ErrNotFound) {
 				return
 			}
-		}
-		entry := resp.Entry
-		if entry == nil {
-			continue
-		}
-		// 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.
-		if !inclusiveStartFrom && marker != "" && entry.Name == marker {
-			continue
+			err = fmt.Errorf("list entries %+v: %w", request, listErr)
+			return
 		}
 
-		if cursor.maxKeys <= 0 {
-			cursor.isTruncated = true
-			break
-		}
-
-		// Set nextMarker only when we have quota to process this entry
-		nextMarker = entry.Name
-		// Track whether this entry is the exact directory targeted by a trailing-slash prefix
-		// (e.g., prefix "foo" from original prefix "foo/"). After recursing into this directory,
-		// we must stop processing siblings to avoid matching unrelated entries like "foo1000".
-		matchedPrefixDir := cursor.prefixEndsOnDelimiter && entry.Name == prefix && entry.IsDirectory
-		if cursor.prefixEndsOnDelimiter {
-			if entry.Name == prefix && entry.IsDirectory {
-				if delimiter != "/" {
-					cursor.prefixEndsOnDelimiter = false
-				}
-			} else {
-				continue
-			}
-		}
-		if entry.IsDirectory {
-			// glog.V(4).Infof("List Dir Entries %s, file: %s, maxKeys %d", dir, entry.Name, cursor.maxKeys)
-			if entry.Name == s3_constants.MultipartUploadsFolder { // FIXME no need to apply to all directories. this extra also affects maxKeys
-				continue
-			}
-
-			// Process .versions directories immediately to create logical versioned object entries
-			// These directories are never traversed (we continue here), so each is only encountered once
-			if strings.HasSuffix(entry.Name, s3_constants.VersionsFolder) {
-				// Extract object name from .versions directory name
-				baseObjectName := strings.TrimSuffix(entry.Name, s3_constants.VersionsFolder)
-				// Construct full object path relative to bucket
-				bucketFullPath := s3a.bucketDir(bucket)
-				bucketRelativePath := strings.TrimPrefix(dir, bucketFullPath)
-				bucketRelativePath = strings.TrimPrefix(bucketRelativePath, "/")
-				var fullObjectPath string
-				if bucketRelativePath == "" {
-					fullObjectPath = baseObjectName
+		var entriesReceived uint32
+		var lastEntryName string
+		for {
+			resp, recvErr := stream.Recv()
+			if recvErr != nil {
+				if recvErr == io.EOF {
+					break
 				} else {
-					fullObjectPath = bucketRelativePath + "/" + baseObjectName
-				}
-				// Use metadata from the already-fetched .versions directory entry
-				if latestVersionEntry, err := s3a.getLatestVersionEntryFromDirectoryEntry(bucket, fullObjectPath, entry); err == nil {
-					eachEntryFn(dir, latestVersionEntry)
-				} else if !errors.Is(err, ErrDeleteMarker) {
-					// Log unexpected errors (delete markers are expected)
-					glog.V(2).Infof("Skipping versioned object %s due to error: %v", fullObjectPath, err)
+					err = fmt.Errorf("iterating entries %+v: %v", request, recvErr)
+					return
 				}
+			}
+			entry := resp.Entry
+			if entry == nil {
+				continue
+			}
+			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.
+			if !inclusiveStartFrom && marker != "" && entry.Name == marker {
 				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
+			if cursor.maxKeys <= 0 {
+				cursor.isTruncated = true
+				break
+			}
+
+			// Set nextMarker only when we have quota to process this entry
+			nextMarker = entry.Name
+			// Track whether this entry is the exact directory targeted by a trailing-slash prefix
+			// (e.g., prefix "foo" from original prefix "foo/"). After recursing into this directory,
+			// we must stop processing siblings to avoid matching unrelated entries like "foo1000".
+			matchedPrefixDir := cursor.prefixEndsOnDelimiter && entry.Name == prefix && entry.IsDirectory
+			if cursor.prefixEndsOnDelimiter {
+				if entry.Name == prefix && entry.IsDirectory {
+					if delimiter != "/" {
+						cursor.prefixEndsOnDelimiter = false
+					}
+				} else {
+					continue
 				}
-				isKeyObject := entry.IsDirectoryKeyObject()
-				if isKeyObject {
-					// Directory key objects (created via PutObject with trailing "/")
-					// must appear as regular keys in recursive listing mode.
+			}
+			if entry.IsDirectory {
+				// glog.V(4).Infof("List Dir Entries %s, file: %s, maxKeys %d", dir, entry.Name, cursor.maxKeys)
+				if entry.Name == s3_constants.MultipartUploadsFolder { // FIXME no need to apply to all directories. this extra also affects maxKeys
+					continue
+				}
+
+				// Process .versions directories immediately to create logical versioned object entries
+				// These directories are never traversed (we continue here), so each is only encountered once
+				if strings.HasSuffix(entry.Name, s3_constants.VersionsFolder) {
+					// Extract object name from .versions directory name
+					baseObjectName := strings.TrimSuffix(entry.Name, s3_constants.VersionsFolder)
+					// Construct full object path relative to bucket
+					bucketFullPath := s3a.bucketDir(bucket)
+					bucketRelativePath := strings.TrimPrefix(dir, bucketFullPath)
+					bucketRelativePath = strings.TrimPrefix(bucketRelativePath, "/")
+					var fullObjectPath string
+					if bucketRelativePath == "" {
+						fullObjectPath = baseObjectName
+					} else {
+						fullObjectPath = bucketRelativePath + "/" + baseObjectName
+					}
+					// Use metadata from the already-fetched .versions directory entry
+					if latestVersionEntry, err := s3a.getLatestVersionEntryFromDirectoryEntry(bucket, fullObjectPath, entry); err == nil {
+						eachEntryFn(dir, latestVersionEntry)
+					} else if !errors.Is(err, ErrDeleteMarker) {
+						// Log unexpected errors (delete markers are expected)
+						glog.V(2).Infof("Skipping versioned object %s due to error: %v", fullObjectPath, err)
+					}
+					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. Surface it as a directory marker for that explicit probe,
+					// identical to a directory created via PutObject with a trailing "/", so
+					// tools like hadoop-aws can find it. Plain listings are left untouched, so
+					// empty directories left behind by deleted objects are not shown as keys.
+					if explicitDirProbe && !isKeyObject && !childEmitted && !cursor.isTruncated && entry.Attributes != nil {
+						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 {
 					eachEntryFn(dir, entry)
 				}
-				// Recurse into subdirectory to list any children, noting whether the
-				// subtree produced any entries.
-				childEmitted := false
-				subNextMarker, subErr := s3a.doListFilerEntries(client, dir+"/"+entry.Name, "", cursor, "", delimiter, false, bucket, 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. Surface it as a directory marker for that explicit probe,
-				// identical to a directory created via PutObject with a trailing "/", so
-				// tools like hadoop-aws can find it. Plain listings are left untouched, so
-				// empty directories left behind by deleted objects are not shown as keys.
-				if explicitDirProbe && !isKeyObject && !childEmitted && !cursor.isTruncated && entry.Attributes != nil {
-					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 {
 				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
 			}
-		} else {
-			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
-		}
-	}
 
-	// Versioned directories are processed above (lines 524-546)
-	return
+		if cursor.isTruncated || entriesReceived < request.Limit {
+			return
+		}
+		marker = lastEntryName
+		inclusiveStartFrom = false
+	}
 }
 
 func getListObjectsV2Args(values url.Values) (prefix, startAfter, delimiter string, token OptionalString, encodingTypeUrl bool, fetchOwner bool, maxkeys uint16, allowUnordered bool, errCode s3err.ErrorCode) {
diff --git a/weed/s3api/s3api_object_handlers_list_directory_test.go b/weed/s3api/s3api_object_handlers_list_directory_test.go
index ee7623a2a..e4e65500a 100644
--- a/weed/s3api/s3api_object_handlers_list_directory_test.go
+++ b/weed/s3api/s3api_object_handlers_list_directory_test.go
@@ -1,6 +1,7 @@
 package s3api
 
 import (
+	"context"
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -38,14 +39,14 @@ func TestDirectoryListedAsCommonPrefix(t *testing.T) {
 	var seenFiles []string
 
 	_, err := s3a.doListFilerEntries(
+		context.Background(),
 		client,
-		"/buckets/xoa-bucket/xo-vm-backups/data",
-		"", // prefix
+		listDirectoryRequest{
+			dir:       "/buckets/xoa-bucket/xo-vm-backups/data",
+			delimiter: "/", // should yield directory for CommonPrefix processing
+			bucket:    "xoa-bucket",
+		},
 		cursor,
-		"",  // marker
-		"/", // delimiter="/" - should yield directory for CommonPrefix processing
-		false,
-		"xoa-bucket",
 		func(dir string, entry *filer_pb.Entry) {
 			if entry.IsDirectory {
 				seenDirs = append(seenDirs, entry.Name)
@@ -88,7 +89,7 @@ func TestEmptyDirectorySurfacedAsMarker(t *testing.T) {
 	// Mirrors the getFileStatus probe: prefix "logs/" with delimiter "/".
 	cursor := &ListingCursor{maxKeys: 1000, prefixEndsOnDelimiter: true}
 	var seen []*filer_pb.Entry
-	_, err := s3a.doListFilerEntries(client, "/buckets/test", "logs", cursor, "", "/", false, "test",
+	_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test", prefix: "logs", delimiter: "/", bucket: "test"}, cursor,
 		func(dir string, entry *filer_pb.Entry) {
 			seen = append(seen, entry)
 		})
@@ -129,7 +130,7 @@ func TestNonEmptyDirectoryGetsNoPhantomMarker(t *testing.T) {
 	// Recursive listing under the "data/" prefix.
 	cursor := &ListingCursor{maxKeys: 1000, prefixEndsOnDelimiter: true}
 	var seen []string
-	_, err := s3a.doListFilerEntries(client, "/buckets/test", "data", cursor, "", "", false, "test",
+	_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test", prefix: "data", bucket: "test"}, cursor,
 		func(dir string, entry *filer_pb.Entry) {
 			seen = append(seen, entry.Name)
 		})
@@ -173,7 +174,7 @@ func TestEmptyDirectoryHiddenInFlatListing(t *testing.T) {
 	// Plain flat listing: no prefix, no delimiter.
 	cursor := &ListingCursor{maxKeys: 1000}
 	var seen []string
-	_, err := s3a.doListFilerEntries(client, "/buckets/test", "", cursor, "", "", false, "test",
+	_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test", bucket: "test"}, cursor,
 		func(dir string, entry *filer_pb.Entry) {
 			seen = append(seen, entry.Name)
 		})
diff --git a/weed/s3api/s3api_object_handlers_list_test.go b/weed/s3api/s3api_object_handlers_list_test.go
index ffbe81c80..8d5d990cd 100644
--- a/weed/s3api/s3api_object_handlers_list_test.go
+++ b/weed/s3api/s3api_object_handlers_list_test.go
@@ -44,8 +44,7 @@ type testFilerClient struct {
 
 func (c *testFilerClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) {
 	entries := c.entriesByDir[in.Directory]
-	// Simplified mock: implements basic prefix filtering but ignores Limit, StartFromFileName, and InclusiveStartFrom
-	// to keep test logic focused. Prefix "/" is treated as no filter for bucket root compatibility.
+	// Simplified mock: prefix "/" is treated as no filter for bucket root compatibility.
 	if in.Prefix != "" && in.Prefix != "/" {
 		filtered := make([]*filer_pb.Entry, 0)
 		for _, e := range entries {
@@ -56,7 +55,16 @@ func (c *testFilerClient) ListEntries(ctx context.Context, in *filer_pb.ListEntr
 		entries = filtered
 	}
 
-	// Respect Limit
+	if in.StartFromFileName != "" {
+		filtered := make([]*filer_pb.Entry, 0)
+		for _, e := range entries {
+			if e.Name > in.StartFromFileName || (in.InclusiveStartFrom && e.Name == in.StartFromFileName) {
+				filtered = append(filtered, e)
+			}
+		}
+		entries = filtered
+	}
+
 	if in.Limit > 0 && int(in.Limit) < len(entries) {
 		entries = entries[:in.Limit]
 	}
@@ -300,7 +308,7 @@ func TestDoListFilerEntries_BucketRootPrefixSlashDelimiterSlash_ListsDirectories
 
 	cursor := &ListingCursor{maxKeys: 1000}
 	seen := make([]string, 0)
-	_, err := s3a.doListFilerEntries(client, "/buckets/test-bucket", "/", cursor, "", "/", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
+	_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test-bucket", prefix: "/", delimiter: "/", bucket: "test-bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
 		if entry.IsDirectory {
 			seen = append(seen, entry.Name)
 		}
@@ -322,7 +330,7 @@ func TestDoListFilerEntries_ExclusiveStartSkipsMarkerEcho(t *testing.T) {
 
 	cursor := &ListingCursor{maxKeys: 1000}
 	var seen []string
-	nextMarker, err := s3a.doListFilerEntries(client, "/buckets/test-bucket", "", cursor, "test.txt", "", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
+	nextMarker, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test-bucket", marker: "test.txt", bucket: "test-bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
 		seen = append(seen, entry.Name)
 	})
 
@@ -346,7 +354,7 @@ func TestDoListFilerEntries_ExclusiveStartSkipsMarkerEchoWithSubsequentEntries(t
 
 	cursor := &ListingCursor{maxKeys: 1000}
 	var seen []string
-	nextMarker, err := s3a.doListFilerEntries(client, "/buckets/test-bucket", "", cursor, "test.txt", "", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
+	nextMarker, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test-bucket", marker: "test.txt", bucket: "test-bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
 		seen = append(seen, entry.Name)
 	})
 
@@ -355,6 +363,116 @@ func TestDoListFilerEntries_ExclusiveStartSkipsMarkerEchoWithSubsequentEntries(t
 	assert.Equal(t, "zebra.txt", nextMarker)
 }
 
+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.
+	s3a := &S3ApiServer{}
+	client := &testFilerClient{
+		entriesByDir: map[string][]*filer_pb.Entry{
+			"/buckets/test-bucket": {
+				{Name: "00-empty-1", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+				{Name: "00-empty-2", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+				{Name: "00-empty-3", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+				{Name: "00-empty-4", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+				{Name: "00-empty-5", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+				{Name: "zz-real", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+			},
+			"/buckets/test-bucket/zz-real": {
+				{Name: "file.log", Attributes: &filer_pb.FuseAttributes{}},
+			},
+		},
+	}
+
+	cursor := &ListingCursor{maxKeys: 1}
+	var seen []string
+	_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test-bucket", bucket: "test-bucket"}, 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/test-bucket/zz-real/file.log"}, seen)
+	assert.False(t, cursor.isTruncated)
+}
+
+func TestDoListFilerEntries_EmptyDirectoriesInSubdirectoryDoNotHideSiblings(t *testing.T) {
+	// The recursion into "logs" gets its own listing window; empty subdirectories
+	// filling that window must not skip the object behind them.
+	s3a := &S3ApiServer{}
+	client := &testFilerClient{
+		entriesByDir: map[string][]*filer_pb.Entry{
+			"/buckets/test-bucket": {
+				{Name: "logs", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+			},
+			"/buckets/test-bucket/logs": {
+				{Name: "00-empty-1", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+				{Name: "00-empty-2", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+				{Name: "00-empty-3", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+				{Name: "00-empty-4", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+				{Name: "00-empty-5", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+				{Name: "zz.log", Attributes: &filer_pb.FuseAttributes{}},
+			},
+		},
+	}
+
+	cursor := &ListingCursor{maxKeys: 1}
+	var seen []string
+	_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test-bucket", bucket: "test-bucket"}, 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/test-bucket/logs/zz.log"}, seen)
+	assert.False(t, cursor.isTruncated)
+}
+
+func TestDoListFilerEntries_TruncationAcrossEmptyDirectories(t *testing.T) {
+	// Two real objects separated by empty directories: the first fills the quota
+	// and the listing must report truncation with a resumable marker.
+	s3a := &S3ApiServer{}
+	client := &testFilerClient{
+		entriesByDir: map[string][]*filer_pb.Entry{
+			"/buckets/test-bucket": {
+				{Name: "00-empty-1", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+				{Name: "00-empty-2", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+				{Name: "00-empty-3", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+				{Name: "mm-real", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+				{Name: "pp-empty", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+				{Name: "zz-real", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+			},
+			"/buckets/test-bucket/mm-real": {
+				{Name: "file.log", Attributes: &filer_pb.FuseAttributes{}},
+			},
+			"/buckets/test-bucket/zz-real": {
+				{Name: "file.log", Attributes: &filer_pb.FuseAttributes{}},
+			},
+		},
+	}
+
+	cursor := &ListingCursor{maxKeys: 1}
+	var seen []string
+	nextMarker, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test-bucket", bucket: "test-bucket"}, 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/test-bucket/mm-real/file.log"}, seen)
+	assert.True(t, cursor.isTruncated)
+
+	cursor = &ListingCursor{maxKeys: 1}
+	seen = nil
+	_, err = s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test-bucket", marker: nextMarker, bucket: "test-bucket"}, 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/test-bucket/zz-real/file.log"}, seen)
+	assert.False(t, cursor.isTruncated)
+}
+
 func TestAllowUnorderedWithDelimiterValidation(t *testing.T) {
 	t.Run("should return error when allow-unordered=true and delimiter are both present", func(t *testing.T) {
 		// Create a request with both allow-unordered=true and delimiter
@@ -765,7 +883,7 @@ func TestListObjectsV2_Regression(t *testing.T) {
 	// Call doListFilerEntries directly to unit test listing logic in isolation,
 	// simulating parameters passed from listFilerEntries for prefix "reports/".
 
-	_, err := s3a.doListFilerEntries(client, "/buckets/reports", "reports", cursor, "", "", false, "reports", func(dir string, entry *filer_pb.Entry) {
+	_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/reports", prefix: "reports", bucket: "reports"}, cursor, func(dir string, entry *filer_pb.Entry) {
 		if !entry.IsDirectory {
 			results = append(results, entry.Name)
 		}
@@ -803,7 +921,7 @@ func TestListObjectsV2_Regression_Sorting(t *testing.T) {
 	// Without the fix, Limit=1 would cause the lister to stop after "reports-archive",
 	// missing the intended "reports" directory.
 
-	_, err := s3a.doListFilerEntries(client, "/buckets/reports", "reports", cursor, "", "", false, "reports", func(dir string, entry *filer_pb.Entry) {
+	_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/reports", prefix: "reports", bucket: "reports"}, cursor, func(dir string, entry *filer_pb.Entry) {
 		if !entry.IsDirectory {
 			results = append(results, entry.Name)
 		}
@@ -846,7 +964,7 @@ func TestListObjectsV2_PrefixEndingWithSlash_DoesNotMatchSiblings(t *testing.T)
 	cursor := &ListingCursor{maxKeys: 1000, prefixEndsOnDelimiter: true}
 	var results []string
 
-	_, err := s3a.doListFilerEntries(client, "/buckets/bucket/path/to/list", "1", cursor, "", "", false, "bucket", func(dir string, entry *filer_pb.Entry) {
+	_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/bucket/path/to/list", prefix: "1", bucket: "bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
 		if !entry.IsDirectory {
 			results = append(results, entry.Name)
 		}
@@ -878,7 +996,7 @@ func TestListObjectsV2_PrefixEndingWithSlash_WithDelimiter(t *testing.T) {
 	cursor := &ListingCursor{maxKeys: 1000, prefixEndsOnDelimiter: true}
 	var results []string
 
-	_, err := s3a.doListFilerEntries(client, "/buckets/bucket/path/to/list", "1", cursor, "", "/", false, "bucket", func(dir string, entry *filer_pb.Entry) {
+	_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/bucket/path/to/list", prefix: "1", delimiter: "/", bucket: "bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
 		results = append(results, entry.Name)
 	})
 
diff --git a/weed/s3api/s3api_object_handlers_list_versioned_test.go b/weed/s3api/s3api_object_handlers_list_versioned_test.go
index fd2695e03..2bcbfa269 100644
--- a/weed/s3api/s3api_object_handlers_list_versioned_test.go
+++ b/weed/s3api/s3api_object_handlers_list_versioned_test.go
@@ -136,7 +136,7 @@ func TestListObjectsWithVersionedObjects(t *testing.T) {
 			commonPrefixes := []PrefixEntry{}
 			bucketPrefix := fmt.Sprintf("%s/%s/", s3a.option.BucketsPath, tt.bucket)
 
-			_, err := s3a.doListFilerEntries(filerClient, bucketPrefix[:len(bucketPrefix)-1], tt.prefix, cursor, "", tt.delimiter, false, tt.bucket, func(dir string, entry *filer_pb.Entry) {
+			_, err := s3a.doListFilerEntries(context.Background(), filerClient, listDirectoryRequest{dir: bucketPrefix[:len(bucketPrefix)-1], prefix: tt.prefix, delimiter: tt.delimiter, bucket: tt.bucket}, cursor, func(dir string, entry *filer_pb.Entry) {
 				if cursor.maxKeys <= 0 {
 					return
 				}
@@ -237,7 +237,7 @@ func TestVersionedObjectsNoDuplication(t *testing.T) {
 
 	cursor := &ListingCursor{maxKeys: uint16(1000)}
 	contents := []ListEntry{}
-	_, err := s3a.doListFilerEntries(filerClient, "/buckets/test-bucket", "", cursor, "", "", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
+	_, err := s3a.doListFilerEntries(context.Background(), filerClient, listDirectoryRequest{dir: "/buckets/test-bucket", bucket: "test-bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
 		if cursor.maxKeys <= 0 {
 			return
 		}
@@ -295,7 +295,7 @@ func TestVersionedObjectsWithDeleteMarker(t *testing.T) {
 
 	cursor := &ListingCursor{maxKeys: uint16(1000)}
 	contents := []ListEntry{}
-	_, err := s3a.doListFilerEntries(filerClient, "/buckets/test-bucket", "", cursor, "", "", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
+	_, err := s3a.doListFilerEntries(context.Background(), filerClient, listDirectoryRequest{dir: "/buckets/test-bucket", bucket: "test-bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
 		if cursor.maxKeys <= 0 {
 			return
 		}
@@ -344,7 +344,7 @@ func TestVersionedObjectsMaxKeys(t *testing.T) {
 
 	cursor := &ListingCursor{maxKeys: uint16(3)}
 	contents := []ListEntry{}
-	_, err := s3a.doListFilerEntries(filerClient, "/buckets/test-bucket", "", cursor, "", "", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
+	_, err := s3a.doListFilerEntries(context.Background(), filerClient, listDirectoryRequest{dir: "/buckets/test-bucket", bucket: "test-bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
 		if cursor.maxKeys <= 0 {
 			return
 		}
@@ -406,7 +406,7 @@ func TestVersionsDirectoryNotTraversed(t *testing.T) {
 
 	cursor := &ListingCursor{maxKeys: uint16(1000)}
 	contents := []ListEntry{}
-	_, err := s3a.doListFilerEntries(customClient, "/buckets/test-bucket", "", cursor, "", "", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
+	_, err := s3a.doListFilerEntries(context.Background(), customClient, listDirectoryRequest{dir: "/buckets/test-bucket", bucket: "test-bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
 		if cursor.maxKeys <= 0 {
 			return
 		}