fix: lex sort order of listobjects backend.Walk

The original Walk function used the native WalkDir and ReadDir which did not
guarantee lexicographic ordering of results for cases where including directory
slash changes the sort order. This caused incorrect paginated responses because
S3 APIs require strict lexicographic ordering where directories with trailing
slashes sort correctly relative to files. For example, dir1/a.b/ must come
before dir1/a/ in the results, but fs.WalkDir was returning them in filesystem
sort order which reversed the order due to not taking in account the trailing
"/".

This also lead to cases of continuous looping of paginated listobjects results
when the marker was set out of order from the expected results.

To address this fundamental ordering issue, the entire directory traversal
mechanism was replaced with a custom lexicographic sorting approach. The new
implementation reads each directory's contents using ReadDir, then sorts the
entries using custom sort keys that append trailing slashes to directory paths.
This ensures that dir1/a.b/ correctly sorts before dir1/a/, as well as other
similar failing cases,  according to ASCII character ordering rules.

Fixes #1283

backend/walk.go

@@ -19,8 +19,11 @@ import (
 	"errors"
 	"fmt"
 	"io/fs"
+	"slices"
+	"sort"
 	"strings"
 	"syscall"
+	"time"
 
 	"github.com/aws/aws-sdk-go-v2/service/s3/types"
 	"github.com/versity/versitygw/s3response"
@@ -37,6 +40,69 @@ type GetObjFunc func(path string, d fs.DirEntry) (s3response.Object, error)
 
 var ErrSkipObj = errors.New("skip this object")
 
+var (
+	pathSeparator = "/"
+	pathDot       = "."
+)
+
+// fakeDirEntry implements fs.DirEntry for implied directories
+type fakeDirEntry struct {
+	name  string
+	isDir bool
+}
+
+func (f *fakeDirEntry) Name() string {
+	return f.name
+}
+
+func (f *fakeDirEntry) IsDir() bool {
+	return f.isDir
+}
+
+func (f *fakeDirEntry) Type() fs.FileMode {
+	if f.isDir {
+		return fs.ModeDir
+	}
+	return 0
+}
+
+func (f *fakeDirEntry) Info() (fs.FileInfo, error) {
+	return &fakeDirInfo{name: f.name, isDir: f.isDir}, nil
+}
+
+// fakeDirInfo implements fs.FileInfo for implied directories
+type fakeDirInfo struct {
+	name  string
+	isDir bool
+}
+
+func (f *fakeDirInfo) Name() string {
+	return f.name
+}
+
+func (f *fakeDirInfo) Size() int64 {
+	return 0
+}
+
+func (f *fakeDirInfo) Mode() fs.FileMode {
+	if f.isDir {
+		return fs.ModeDir
+	}
+	return 0
+}
+
+func (f *fakeDirInfo) ModTime() time.Time {
+	return time.Time{} // Return zero time for fake directories
+}
+
+func (f *fakeDirInfo) IsDir() bool {
+	return f.isDir
+}
+
+func (f *fakeDirInfo) Sys() any {
+	return nil
+}
+
 // map to store object common prefixes
 type cpMap map[string]int
 
@@ -65,247 +131,477 @@ func (c cpMap) CpArray() []types.CommonPrefix {
 	return commonPrefixes
 }
 
+// walkState holds the state needed during directory traversal
+type walkState struct {
+	ctx        context.Context
+	fileSystem fs.FS
+	prefix     string
+	delimiter  string
+	marker     string
+	max        int32
+	getObj     GetObjFunc
+	skipdirs   []string
+
+	// Mutable state
+	cpmap      cpMap
+	objects    []s3response.Object
+	pastMarker bool
+	pastMax    bool
+	newMarker  string
+	truncated  bool
+}
+
+// entryWithSortKey represents a directory entry with its lexicographic sort key
+type entryWithSortKey struct {
+	d       fs.DirEntry
+	path    string
+	sortKey string
+}
+
+// shouldSkipDirectory checks if a directory should be skipped based on prefix matching
+func shouldSkipDirectory(dirPath, prefix string) bool {
+	if prefix == "" {
+		return false
+	}
+	return !strings.HasPrefix(dirPath, prefix) && !strings.HasPrefix(prefix, dirPath)
+}
+
+// constructPath builds the full path for a directory entry
+func constructPath(dir, entryName string) string {
+	if dir == pathDot {
+		return entryName
+	}
+	return dir + pathSeparator + entryName
+}
+
+// createSortKey creates a lexicographic sort key for an entry
+func createSortKey(path string, isDir bool) string {
+	if isDir {
+		return path + pathSeparator
+	}
+	return path
+}
+
+// buildSortedEntries reads directory entries and sorts them lexicographically
+func buildSortedEntries(fileSystem fs.FS, dir string, skipdirs []string) ([]entryWithSortKey, error) {
+	entries, err := fs.ReadDir(fileSystem, dir)
+	if err != nil {
+		return nil, err
+	}
+
+	var entriesWithKeys []entryWithSortKey
+	for _, d := range entries {
+		if slices.Contains(skipdirs, d.Name()) {
+			continue
+		}
+
+		path := constructPath(dir, d.Name())
+		sortKey := createSortKey(path, d.IsDir())
+
+		entriesWithKeys = append(entriesWithKeys, entryWithSortKey{
+			d:       d,
+			path:    path,
+			sortKey: sortKey,
+		})
+	}
+
+	// Sort by the sort key to ensure proper lexicographic ordering
+	sort.Slice(entriesWithKeys, func(i, j int) bool {
+		return entriesWithKeys[i].sortKey < entriesWithKeys[j].sortKey
+	})
+
+	return entriesWithKeys, nil
+}
+
+// checkLimits returns true if we've hit our limits and should stop processing
+func (w *walkState) checkLimits() bool {
+	return w.pastMax || w.ctx.Err() != nil
+}
+
+// addObject adds an object to the results and checks if limits are reached
+func (w *walkState) addObject(obj s3response.Object, path string) bool {
+	w.objects = append(w.objects, obj)
+	if len(w.objects)+w.cpmap.Len() >= int(w.max) {
+		w.newMarker = path
+		w.pastMax = true
+		// Don't set truncated here - wait until we know there are more items
+		return true
+	}
+	return false
+}
+
+// addCommonPrefix adds a common prefix and checks if limits are reached
+func (w *walkState) addCommonPrefix(cpref string) bool {
+	w.cpmap.Add(cpref)
+	if len(w.objects)+w.cpmap.Len() >= int(w.max) {
+		w.newMarker = cpref
+		w.pastMax = true
+		// Don't set truncated here - wait until we know there are more items
+		return true
+	}
+	return false
+}
+
+// isBeforeMarker checks if a path is before the current marker
+func (w *walkState) isBeforeMarker(path string) bool {
+	return !w.pastMarker && path < w.marker
+}
+
+// isAtMarker checks if a path matches the current marker
+func (w *walkState) isAtMarker(path string) bool {
+	return !w.pastMarker && path == w.marker
+}
+
 // Walk walks the supplied fs.FS and returns results compatible with list
 // objects responses
 func Walk(ctx context.Context, fileSystem fs.FS, prefix, delimiter, marker string, max int32, getObj GetObjFunc, skipdirs []string) (WalkResults, error) {
-	cpmap := cpMap{}
-	var objects []s3response.Object
-
-	// if max is 0, it should return empty non-truncated result
+	// Early return for zero max
 	if max == 0 {
-		return WalkResults{
-			Truncated: false,
-		}, nil
+		return WalkResults{Truncated: false}, nil
 	}
 
-	var pastMarker bool
-	if marker == "" {
-		pastMarker = true
+	state := &walkState{
+		ctx:        ctx,
+		fileSystem: fileSystem,
+		prefix:     prefix,
+		delimiter:  delimiter,
+		marker:     marker,
+		max:        max,
+		getObj:     getObj,
+		skipdirs:   skipdirs,
+		cpmap:      cpMap{},
+		pastMarker: marker == "",
 	}
 
-	var pastMax bool
-	var newMarker string
-	var truncated bool
-
-	root := "."
-	if strings.Contains(prefix, "/") {
-		idx := strings.LastIndex(prefix, "/")
-		if idx > 0 {
+	root := pathDot
+	if strings.Contains(prefix, pathSeparator) {
+		if idx := strings.LastIndex(prefix, pathSeparator); idx > 0 {
 			root = prefix[:idx]
 		}
 	}
 
-	err := fs.WalkDir(fileSystem, root, func(path string, d fs.DirEntry, err error) error {
+	// Special case: if prefix ends with delimiter and refers to an empty single directory,
+	// we need to start from parent to find the directory as an entry
+	if delimiter != "" && prefix != "" && strings.HasSuffix(prefix, delimiter) {
+		prefixDir := strings.TrimSuffix(prefix, delimiter)
+		if !strings.Contains(prefixDir, delimiter) {
+			// Check if this directory exists and is empty
+			if entries, err := fs.ReadDir(fileSystem, prefixDir); err == nil && len(entries) == 0 {
+				// Single-level empty directory like "a" with prefix "a/"
+				root = pathDot
+			}
+		}
+	}
+
+	// Handle special case for prefix directories in non-delimiter mode
+	if delimiter == "" && prefix != "" && strings.HasSuffix(prefix, pathSeparator) {
+		err := state.handlePrefixDirectory(prefix)
 		if err != nil {
-			return err
-		}
-		if ctx.Err() != nil {
-			return ctx.Err()
-		}
-		// Ignore the root directory
-		if path == "." {
-			return nil
-		}
-		if contains(d.Name(), skipdirs) {
-			return fs.SkipDir
+			return WalkResults{}, err
 		}
+	}
 
-		// After this point, return skipflag instead of nil
-		// so we can skip a directory without an early return
-		var skipflag error
-		if d.IsDir() {
-			// If prefix is defined and the directory does not match prefix,
-			// do not descend into the directory because nothing will
-			// match this prefix. Make sure to append the / at the end of
-			// directories since this is implied as a directory path name.
-			// If path is a prefix of prefix, then path could still be
-			// building to match. So only skip if path isn't a prefix of prefix
-			// and prefix isn't a prefix of path.
-			if prefix != "" &&
-				!strings.HasPrefix(path+"/", prefix) &&
-				!strings.HasPrefix(prefix, path+"/") {
-				return fs.SkipDir
-			}
-
-			// Don't recurse into subdirectories which contain the delimiter
-			// after reaching the prefix
-			if delimiter != "" &&
-				strings.HasPrefix(path+"/", prefix) &&
-				strings.Contains(strings.TrimPrefix(path+"/", prefix), delimiter) {
-				skipflag = fs.SkipDir
-			} else {
-				if delimiter == "" {
-					dirobj, err := getObj(path+"/", d)
-					if err == ErrSkipObj {
-						return skipflag
-					}
-					if err != nil {
-						return fmt.Errorf("directory to object %q: %w", path, err)
-					}
-					if pastMax {
-						truncated = true
-						return fs.SkipAll
-					}
-					objects = append(objects, dirobj)
-					if (len(objects) + cpmap.Len()) == int(max) {
-						newMarker = path
-						pastMax = true
-					}
-
-					return skipflag
-				}
-
-				// TODO: can we do better here rather than a second readdir
-				// per directory?
-				ents, err := fs.ReadDir(fileSystem, path)
-				if err != nil {
-					return fmt.Errorf("readdir %q: %w", path, err)
-				}
-
-				if len(ents) != 0 {
-					return skipflag
-				}
-			}
-			path += "/"
-		}
-
-		if !pastMarker {
-			if path == marker {
-				pastMarker = true
-				return skipflag
-			}
-			if path < marker {
-				return skipflag
-			}
-		}
-
-		// If object doesn't have prefix, don't include in results.
-		if prefix != "" && !strings.HasPrefix(path, prefix) {
-			return skipflag
-		}
-
-		if delimiter == "" {
-			// If no delimiter specified, then all files with matching
-			// prefix are included in results
-			obj, err := getObj(path, d)
-			if err == ErrSkipObj {
-				return skipflag
-			}
-			if err != nil {
-				return fmt.Errorf("file to object %q: %w", path, err)
-			}
-			if pastMax {
-				truncated = true
-				return fs.SkipAll
-			}
-
-			objects = append(objects, obj)
-
-			if (len(objects) + cpmap.Len()) == int(max) {
-				newMarker = path
-				pastMax = true
-			}
-
-			return skipflag
-		}
-
-		// Since delimiter is specified, we only want results that
-		// do not contain the delimiter beyond the prefix.  If the
-		// delimiter exists past the prefix, then the substring
-		// between the prefix and delimiter is part of common prefixes.
-		//
-		// For example:
-		// prefix = A/
-		// delimiter = /
-		// and objects:
-		// A/file
-		// A/B/file
-		// B/C
-		// would return:
-		// objects: A/file
-		// common prefix: A/B/
-		//
-		// Note: No objects are included past the common prefix since
-		// these are all rolled up into the common prefix.
-		// Note: The delimiter can be anything, so we have to operate on
-		// the full path without any assumptions on posix directory hierarchy
-		// here.  Usually the delimiter will be "/", but thats not required.
-		suffix := strings.TrimPrefix(path, prefix)
-		before, _, found := strings.Cut(suffix, delimiter)
-		if !found {
-			obj, err := getObj(path, d)
-			if err == ErrSkipObj {
-				return skipflag
-			}
-			if err != nil {
-				return fmt.Errorf("file to object %q: %w", path, err)
-			}
-			if pastMax {
-				truncated = true
-				return fs.SkipAll
-			}
-			objects = append(objects, obj)
-			if (len(objects) + cpmap.Len()) == int(max) {
-				newMarker = path
-				pastMax = true
-			}
-			return skipflag
-		}
-
-		// Common prefixes are a set, so should not have duplicates.
-		// These are abstractly a "directory", so need to include the
-		// delimiter at the end when we add to the map.
-		cprefNoDelim := prefix + before
-		cpref := prefix + before + delimiter
-		if cpref == marker {
-			pastMarker = true
-			return skipflag
-		}
-
-		if marker != "" && strings.HasPrefix(marker, cprefNoDelim) {
-			// skip common prefixes that are before the marker
-			return skipflag
-		}
-
-		if pastMax {
-			truncated = true
-			return fs.SkipAll
-		}
-		cpmap.Add(cpref)
-		if (len(objects) + cpmap.Len()) == int(max) {
-			newMarker = cpref
-			pastMax = true
-		}
-
-		return skipflag
-	})
-	if err != nil {
-		// suppress file not found caused by user's prefix
+	if err := state.walkLexSort(root); err != nil {
 		if errors.Is(err, fs.ErrNotExist) || errors.Is(err, syscall.ENOTDIR) {
 			return WalkResults{}, nil
 		}
 		return WalkResults{}, err
 	}
 
-	if !truncated {
-		newMarker = ""
+	if !state.truncated {
+		state.newMarker = ""
 	}
 
 	return WalkResults{
-		CommonPrefixes: cpmap.CpArray(),
-		Objects:        objects,
-		Truncated:      truncated,
-		NextMarker:     newMarker,
+		CommonPrefixes: state.cpmap.CpArray(),
+		Objects:        state.objects,
+		Truncated:      state.truncated,
+		NextMarker:     state.newMarker,
 	}, nil
 }
 
-func contains(a string, strs []string) bool {
-	for _, s := range strs {
-		if s == a {
-			return true
+// handlePrefixDirectory handles the special case where we need to include the root directory itself
+func (w *walkState) handlePrefixDirectory(prefix string) error {
+	if w.pastMarker || w.marker == "" || prefix <= w.marker {
+		if w.marker == "" || prefix < w.marker || prefix == w.marker {
+			if prefix != w.marker {
+				prefixDir := strings.TrimSuffix(prefix, pathSeparator)
+				dirInfo := &fakeDirEntry{name: prefixDir, isDir: true}
+				dirobj, err := w.getObj(prefix, dirInfo)
+				switch err {
+				case ErrSkipObj:
+					// Skip this directory
+				case nil:
+					if w.addObject(dirobj, prefix) {
+						return nil
+					}
+				default:
+					return fmt.Errorf("prefix directory to object %q: %w",
+						prefix, err)
+				}
+			}
+			if prefix == w.marker {
+				w.pastMarker = true
+			}
 		}
 	}
-	return false
+	return nil
+}
+
+// walkLexSort performs lexicographically ordered directory traversal
+func (w *walkState) walkLexSort(dir string) error {
+	if w.ctx.Err() != nil {
+		return w.ctx.Err()
+	}
+
+	entries, err := buildSortedEntries(w.fileSystem, dir, w.skipdirs)
+	if err != nil {
+		return err
+	}
+
+	for i, entry := range entries {
+		if w.checkLimits() {
+			return w.ctx.Err()
+		}
+
+		if w.pastMax {
+			// We have more entries to process, so mark as truncated
+			w.truncated = true
+			return nil
+		}
+
+		if err := w.processEntry(entry); err != nil {
+			return err
+		}
+
+		// After processing an entry, check if we hit the limit
+		if w.pastMax {
+			// We hit the limit. Check if there are more entries after this one
+			if i+1 < len(entries) {
+				// There are more entries in this directory
+				w.truncated = true
+			} else {
+				// This was the last entry in this directory, but there might be more dirs to recurse into
+				// Let the higher level determine if we're truly done
+			}
+			return nil
+		}
+	}
+
+	return nil
+}
+
+// processEntry processes a single directory entry
+func (w *walkState) processEntry(entry entryWithSortKey) error {
+	path := entry.path
+	d := entry.d
+
+	if d.IsDir() {
+		return w.processDirectory(path, d)
+	}
+	return w.processFile(path, d)
+}
+
+// processDirectory handles directory processing logic
+func (w *walkState) processDirectory(path string, d fs.DirEntry) error {
+	dirPath := path + pathSeparator
+
+	// Check if we should skip this directory based on prefix
+	if shouldSkipDirectory(dirPath, w.prefix) {
+		return nil
+	}
+
+	// Handle marker logic for directories
+	if w.isBeforeMarker(dirPath) {
+		// Before marker - only recurse if marker could be inside
+		if strings.HasPrefix(w.marker, dirPath) {
+			return w.walkLexSort(path)
+		}
+		return nil
+	}
+
+	if w.isAtMarker(dirPath) {
+		// At marker - recurse but don't include as common prefix
+		w.pastMarker = true
+		return w.walkLexSort(path)
+	}
+
+	// Apply prefix filter
+	if w.prefix != "" && !strings.HasPrefix(dirPath, w.prefix) {
+		return w.walkLexSort(path)
+	}
+
+	if w.delimiter != "" {
+		return w.processDirectoryWithDelimiter(path, dirPath)
+	}
+	return w.processDirectoryWithoutDelimiter(path, dirPath, d)
+}
+
+// processDirectoryWithDelimiter handles directory processing when delimiter is specified
+func (w *walkState) processDirectoryWithDelimiter(path, dirPath string) error {
+	// Special case: if the directory path exactly matches the prefix, return it as an object
+	if dirPath == w.prefix {
+		dirobj, err := w.getObj(dirPath, &fakeDirEntry{name: path, isDir: true})
+		switch err {
+		case ErrSkipObj:
+			// Skip but continue to recurse
+		case nil:
+			if w.addObject(dirobj, dirPath) {
+				return nil
+			}
+		default:
+			return fmt.Errorf("directory to object %q: %w", dirPath, err)
+		}
+	}
+
+	// Handle delimiter logic for directories
+	suffix := strings.TrimPrefix(dirPath, w.prefix)
+	before, _, found := strings.Cut(suffix, w.delimiter)
+
+	if found {
+		// Directory creates a common prefix
+		cprefNoDelim := w.prefix + before
+		cpref := w.prefix + before + w.delimiter
+
+		if w.isBeforeMarker(cpref) {
+			if w.marker != "" && strings.HasPrefix(w.marker, cprefNoDelim) {
+				return w.walkLexSort(path)
+			}
+			return nil
+		}
+
+		if w.isAtMarker(cpref) {
+			w.pastMarker = true
+			return w.walkLexSort(path)
+		}
+
+		// Skip if this common prefix is <= marker (for when pastMarker is already true)
+		if w.marker != "" && cpref <= w.marker {
+			return w.walkLexSort(path)
+		}
+
+		if w.pastMax {
+			w.truncated = true
+			return nil
+		}
+
+		if w.addCommonPrefix(cpref) {
+			return nil
+		}
+	}
+
+	return w.walkLexSort(path)
+}
+
+// processDirectoryWithoutDelimiter handles directory processing when no delimiter is specified
+func (w *walkState) processDirectoryWithoutDelimiter(path, dirPath string, d fs.DirEntry) error {
+	// Include directory as object if it matches prefix
+	if w.prefix == "" || strings.HasPrefix(dirPath, w.prefix) {
+		dirobj, err := w.getObj(dirPath, d)
+		switch err {
+		case ErrSkipObj:
+			// Skip but continue to recurse
+		case nil:
+			if w.addObject(dirobj, dirPath) {
+				return nil
+			}
+		default:
+			return fmt.Errorf("directory to object %q: %w", dirPath, err)
+		}
+	}
+
+	return w.walkLexSort(path)
+}
+
+func (w *walkState) processFile(path string, d fs.DirEntry) error {
+	if w.isBeforeMarker(path) {
+		return nil
+	}
+
+	if w.isAtMarker(path) {
+		w.pastMarker = true
+		return nil
+	}
+
+	if w.prefix != "" && !strings.HasPrefix(path, w.prefix) {
+		return nil
+	}
+
+	if w.delimiter != "" {
+		return w.processFileWithDelimiter(path, d)
+	}
+	return w.processFileWithoutDelimiter(path, d)
+}
+
+// processFileWithDelimiter handles file processing when delimiter is specified
+func (w *walkState) processFileWithDelimiter(path string, d fs.DirEntry) error {
+	suffix := strings.TrimPrefix(path, w.prefix)
+	before, _, found := strings.Cut(suffix, w.delimiter)
+
+	if !found {
+		// File doesn't contain delimiter after prefix - include it
+		obj, err := w.getObj(path, d)
+		if err == ErrSkipObj {
+			return nil
+		}
+		if err != nil {
+			return fmt.Errorf("file to object %q: %w", path, err)
+		}
+		if w.pastMax {
+			w.truncated = true
+			return nil
+		}
+
+		if w.addObject(obj, path) {
+			return nil
+		}
+
+		return nil
+	}
+
+	// File contains delimiter after prefix - add to common prefixes
+	cprefNoDelim := w.prefix + before
+	cpref := w.prefix + before + w.delimiter
+
+	if w.isBeforeMarker(cpref) {
+		if w.marker != "" && strings.HasPrefix(w.marker, cprefNoDelim) {
+			return nil
+		}
+		return nil
+	}
+
+	if w.isAtMarker(cpref) {
+		w.pastMarker = true
+		return nil
+	}
+
+	// Skip if this common prefix is <= marker (for when pastMarker is already true)
+	if w.marker != "" && cpref <= w.marker {
+		return nil
+	}
+
+	if w.pastMax {
+		w.truncated = true
+		return nil
+	}
+
+	w.addCommonPrefix(cpref)
+
+	return nil
+}
+
+// processFileWithoutDelimiter handles file processing when no delimiter is specified
+func (w *walkState) processFileWithoutDelimiter(path string, d fs.DirEntry) error {
+	obj, err := w.getObj(path, d)
+	if err == ErrSkipObj {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("file to object %q: %w", path, err)
+	}
+
+	w.addObject(obj, path)
+	return nil
 }
 
 type WalkVersioningResults struct {
@@ -343,7 +639,7 @@ func WalkVersions(ctx context.Context, fileSystem fs.FS, prefix, delimiter, keyM
 
 	pastVersionIdMarker := versionIdMarker == ""
 
-	err := fs.WalkDir(fileSystem, ".", func(path string, d fs.DirEntry, err error) error {
+	err := fs.WalkDir(fileSystem, pathDot, func(path string, d fs.DirEntry, err error) error {
 		if err != nil {
 			return err
 		}
@@ -351,10 +647,10 @@ func WalkVersions(ctx context.Context, fileSystem fs.FS, prefix, delimiter, keyM
 			return ctx.Err()
 		}
 		// Ignore the root directory
-		if path == "." {
+		if path == pathDot {
 			return nil
 		}
-		if contains(d.Name(), skipdirs) {
+		if slices.Contains(skipdirs, d.Name()) {
 			return fs.SkipDir
 		}
 
@@ -376,16 +672,16 @@ func WalkVersions(ctx context.Context, fileSystem fs.FS, prefix, delimiter, keyM
 			// building to match. So only skip if path isn't a prefix of prefix
 			// and prefix isn't a prefix of path.
 			if prefix != "" &&
-				!strings.HasPrefix(path+"/", prefix) &&
-				!strings.HasPrefix(prefix, path+"/") {
+				!strings.HasPrefix(path+pathSeparator, prefix) &&
+				!strings.HasPrefix(prefix, path+pathSeparator) {
 				return fs.SkipDir
 			}
 
 			// Don't recurse into subdirectories when listing with delimiter.
-			if delimiter == "/" &&
-				prefix != path+"/" &&
-				strings.HasPrefix(path+"/", prefix) {
-				cpmap.Add(path + "/")
+			if delimiter == pathSeparator &&
+				prefix != path+pathSeparator &&
+				strings.HasPrefix(path+pathSeparator, prefix) {
+				cpmap.Add(path + pathSeparator)
 				return fs.SkipDir
 			}
 
@@ -455,7 +751,7 @@ func WalkVersions(ctx context.Context, fileSystem fs.FS, prefix, delimiter, keyM
 		// these are all rolled up into the common prefix.
 		// Note: The delimiter can be anything, so we have to operate on
 		// the full path without any assumptions on posix directory hierarchy
-		// here.  Usually the delimiter will be "/", but thats not required.
+		// here.  Usually the delimiter will be pathSeparator, but thats not required.
 		suffix := strings.TrimPrefix(path, prefix)
 		before, _, found := strings.Cut(suffix, delimiter)
 		if !found {

backend/walk_test.go

@@ -112,6 +112,22 @@ func TestWalk(t *testing.T) {
 						}},
 					},
 				},
+				{
+					name:      "max objs",
+					delimiter: "/",
+					prefix:    "photos/2006/February/",
+					maxObjs:   2,
+					expected: backend.WalkResults{
+						Objects: []s3response.Object{
+							{
+								Key: backend.GetPtrFromString("photos/2006/February/sample2.jpg"),
+							},
+							{
+								Key: backend.GetPtrFromString("photos/2006/February/sample3.jpg"),
+							},
+						},
+					},
+				},
 			},
 		},
 		{
@@ -226,7 +242,7 @@ func TestWalk(t *testing.T) {
 				tt.fsys, tc.prefix, tc.delimiter, tc.marker, tc.maxObjs,
 				tt.getobj, []string{})
 			if err != nil {
-				t.Errorf("tc.name: walk: %v", err)
+				t.Errorf("%v: walk: %v", tc.name, err)
 			}
 
 			compareResults(tc.name, res, tc.expected, t)
@@ -376,3 +392,412 @@ func TestWalkStop(t *testing.T) {
 		t.Fatalf("unexpected error: %v", err)
 	}
 }
+
+// TestOrderWalk tests the lexicographic ordering of the object names
+// for the case where readdir sort order of a directory is different
+// than the lexicographic ordering of the full paths. The below has
+// a readdir sort order for dir1/:
+// a, a.b
+// but if you consider the character that comes after a is "/", then
+// the "." should come before "/" in the lexicographic ordering:
+// a.b/, a/
+func TestOrderWalk(t *testing.T) {
+	tests := []walkTest{
+		{
+			fsys: fstest.MapFS{
+				"dir1/a/file1":   {},
+				"dir1/a/file2":   {},
+				"dir1/a/file3":   {},
+				"dir1/a.b/file1": {},
+				"dir1/a.b/file2": {},
+			},
+			getobj: getObj,
+			cases: []testcase{
+				{
+					name:    "order test",
+					maxObjs: 1000,
+					prefix:  "dir1/",
+					expected: backend.WalkResults{
+						Objects: []s3response.Object{
+							{
+								Key: backend.GetPtrFromString("dir1/"),
+							},
+							{
+								Key: backend.GetPtrFromString("dir1/a.b/"),
+							},
+							{
+								Key: backend.GetPtrFromString("dir1/a.b/file1"),
+							},
+							{
+								Key: backend.GetPtrFromString("dir1/a.b/file2"),
+							},
+							{
+								Key: backend.GetPtrFromString("dir1/a/"),
+							},
+							{
+								Key: backend.GetPtrFromString("dir1/a/file1"),
+							},
+							{
+								Key: backend.GetPtrFromString("dir1/a/file2"),
+							},
+							{
+								Key: backend.GetPtrFromString("dir1/a/file3"),
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			fsys: fstest.MapFS{
+				"dir|1/a/file1":   {},
+				"dir|1/a/file2":   {},
+				"dir|1/a/file3":   {},
+				"dir|1/a.b/file1": {},
+				"dir|1/a.b/file2": {},
+			},
+			getobj: getObj,
+			cases: []testcase{
+				{
+					name:      "order test delim",
+					maxObjs:   1000,
+					delimiter: "|",
+					prefix:    "dir|",
+					expected: backend.WalkResults{
+						Objects: []s3response.Object{
+							{
+								Key: backend.GetPtrFromString("dir|1/a.b/file1"),
+							},
+							{
+								Key: backend.GetPtrFromString("dir|1/a.b/file2"),
+							},
+							{
+								Key: backend.GetPtrFromString("dir|1/a/file1"),
+							},
+							{
+								Key: backend.GetPtrFromString("dir|1/a/file2"),
+							},
+							{
+								Key: backend.GetPtrFromString("dir|1/a/file3"),
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			fsys: fstest.MapFS{
+				"a": &fstest.MapFile{Mode: fs.ModeDir},
+			},
+			getobj: getObj,
+			cases: []testcase{
+				{
+					name:      "single dir obj",
+					maxObjs:   1000,
+					delimiter: "/",
+					prefix:    "a",
+					expected: backend.WalkResults{
+						CommonPrefixes: []types.CommonPrefix{
+							{
+								Prefix: backend.GetPtrFromString("a/"),
+							},
+						},
+					},
+				},
+				{
+					name:      "single dir obj",
+					maxObjs:   1000,
+					delimiter: "/",
+					prefix:    "a/",
+					expected: backend.WalkResults{
+						Objects: []s3response.Object{
+							{
+								Key: backend.GetPtrFromString("a/"),
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		for _, tc := range tt.cases {
+			res, err := backend.Walk(context.Background(),
+				tt.fsys, tc.prefix, tc.delimiter, tc.marker, tc.maxObjs,
+				tt.getobj, []string{})
+			if err != nil {
+				t.Errorf("%v: walk: %v", tc.name, err)
+			}
+
+			compareResultsOrdered(tc.name, res, tc.expected, t)
+		}
+	}
+}
+
+type markerTest struct {
+	fsys   fs.FS
+	getobj backend.GetObjFunc
+	cases  []markertestcase
+}
+
+type markertestcase struct {
+	name      string
+	prefix    string
+	delimiter string
+	marker    string
+	maxObjs   int32
+	expected  []backend.WalkResults
+}
+
+func TestMarker(t *testing.T) {
+	tests := []markerTest{
+		{
+			fsys: fstest.MapFS{
+				"dir/sample2.jpg": {},
+				"dir/sample3.jpg": {},
+				"dir/sample4.jpg": {},
+				"dir/sample5.jpg": {},
+			},
+			getobj: getObj,
+			cases: []markertestcase{
+				{
+					name:      "multi page marker",
+					delimiter: "/",
+					prefix:    "dir/",
+					maxObjs:   2,
+					expected: []backend.WalkResults{
+						{
+							Objects: []s3response.Object{
+								{
+									Key: backend.GetPtrFromString("dir/sample2.jpg"),
+								},
+								{
+									Key: backend.GetPtrFromString("dir/sample3.jpg"),
+								},
+							},
+							Truncated: true,
+						},
+						{
+							Objects: []s3response.Object{
+								{
+									Key: backend.GetPtrFromString("dir/sample4.jpg"),
+								},
+								{
+									Key: backend.GetPtrFromString("dir/sample5.jpg"),
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			fsys: fstest.MapFS{
+				"dir1/subdir/file.txt": {},
+				"dir1/subdir.ext":      {},
+				"dir1/subdir1.ext":     {},
+				"dir1/subdir2.ext":     {},
+			},
+			getobj: getObj,
+			cases: []markertestcase{
+				{
+					name:      "integration test case 1",
+					maxObjs:   2,
+					delimiter: "/",
+					prefix:    "dir1/",
+					expected: []backend.WalkResults{
+						{
+							Objects: []s3response.Object{
+								{
+									Key: backend.GetPtrFromString("dir1/subdir.ext"),
+								},
+							},
+							CommonPrefixes: []types.CommonPrefix{
+								{
+									Prefix: backend.GetPtrFromString("dir1/subdir/"),
+								},
+							},
+							Truncated: true,
+						},
+						{
+							Objects: []s3response.Object{
+								{
+									Key: backend.GetPtrFromString("dir1/subdir1.ext"),
+								},
+								{
+									Key: backend.GetPtrFromString("dir1/subdir2.ext"),
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			fsys: fstest.MapFS{
+				"asdf":          {},
+				"boo/bar":       {},
+				"boo/baz/xyzzy": {},
+				"cquux/thud":    {},
+				"cquux/bla":     {},
+			},
+			getobj: getObj,
+			cases: []markertestcase{
+				{
+					name:      "integration test case2",
+					maxObjs:   1,
+					delimiter: "/",
+					marker:    "boo/",
+					expected: []backend.WalkResults{
+						{
+							Objects: []s3response.Object{},
+							CommonPrefixes: []types.CommonPrefix{
+								{
+									Prefix: backend.GetPtrFromString("cquux/"),
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			fsys: fstest.MapFS{
+				"bar": {},
+				"baz": {},
+				"foo": {},
+			},
+			getobj: getObj,
+			cases: []markertestcase{
+				{
+					name:    "exact limit count",
+					maxObjs: 3,
+					expected: []backend.WalkResults{
+						{
+							Objects: []s3response.Object{
+								{
+									Key: backend.GetPtrFromString("bar"),
+								},
+								{
+									Key: backend.GetPtrFromString("baz"),
+								},
+								{
+									Key: backend.GetPtrFromString("foo"),
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			fsys: fstest.MapFS{
+				"d1/f1": {},
+				"d2/f2": {},
+				"d3/f3": {},
+				"d4/f4": {},
+			},
+			getobj: getObj,
+			cases: []markertestcase{
+				{
+					name:      "limited common prefix",
+					maxObjs:   3,
+					delimiter: "/",
+					expected: []backend.WalkResults{
+						{
+							CommonPrefixes: []types.CommonPrefix{
+								{
+									Prefix: backend.GetPtrFromString("d1/"),
+								},
+								{
+									Prefix: backend.GetPtrFromString("d2/"),
+								},
+								{
+									Prefix: backend.GetPtrFromString("d3/"),
+								},
+							},
+							Truncated: true,
+						},
+						{
+							CommonPrefixes: []types.CommonPrefix{
+								{
+									Prefix: backend.GetPtrFromString("d4/"),
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		for _, tc := range tt.cases {
+			marker := tc.marker
+			for i, page := range tc.expected {
+				res, err := backend.Walk(context.Background(),
+					tt.fsys, tc.prefix, tc.delimiter, marker, tc.maxObjs,
+					tt.getobj, []string{})
+				if err != nil {
+					t.Errorf("%v: walk: %v", tc.name, err)
+				}
+				marker = res.NextMarker
+
+				compareResultsOrdered(tc.name, res, page, t)
+
+				if res.Truncated != page.Truncated {
+					t.Errorf("%v page %v expected truncated %v, got %v",
+						tc.name, i, page.Truncated, res.Truncated)
+				}
+			}
+		}
+	}
+}
+
+func compareResultsOrdered(name string, got, wanted backend.WalkResults, t *testing.T) {
+	if !compareObjectsOrdered(got.Objects, wanted.Objects) {
+		t.Errorf("%v: unexpected object, got %v wanted %v",
+			name,
+			printObjects(got.Objects),
+			printObjects(wanted.Objects))
+	}
+	if !comparePrefixesOrdered(got.CommonPrefixes, wanted.CommonPrefixes) {
+		t.Errorf("%v: unexpected prefix, got %v wanted %v",
+			name,
+			printCommonPrefixes(got.CommonPrefixes),
+			printCommonPrefixes(wanted.CommonPrefixes))
+	}
+}
+
+func compareObjectsOrdered(a, b []s3response.Object) bool {
+	if len(a) == 0 && len(b) == 0 {
+		return true
+	}
+	if len(a) != len(b) {
+		return false
+	}
+
+	for i, obj := range a {
+		if *obj.Key != *b[i].Key {
+			return false
+		}
+	}
+	return true
+}
+
+func comparePrefixesOrdered(a, b []types.CommonPrefix) bool {
+	if len(a) == 0 && len(b) == 0 {
+		return true
+	}
+	if len(a) != len(b) {
+		return false
+	}
+
+	for i, cp := range a {
+		if *cp.Prefix != *b[i].Prefix {
+			return false
+		}
+	}
+	return true
+}

tests/integration/tests.go

@@ -1459,7 +1459,7 @@ func CreateBucket_existing_bucket(s *S3Conf) error {
 
 	err = teardown(s, bucket)
 	if err != nil {
-		failF("%v: %v", err)
+		failF("%v: %v", testName, err)
 		return fmt.Errorf("%v: %w", testName, err)
 	}
 	passF(testName)
@@ -1567,7 +1567,7 @@ func CreateBucket_non_default_acl(s *S3Conf) error {
 		{"grt3", "grt3secret", "user"},
 	})
 	if err != nil {
-		failF("%v: %v", err)
+		failF("%v: %v", testName, err)
 		return fmt.Errorf("%v: %w", testName, err)
 	}
 
@@ -1615,7 +1615,7 @@ func CreateBucket_non_default_acl(s *S3Conf) error {
 	})
 	cancel()
 	if err != nil {
-		failF("%v: %v", err)
+		failF("%v: %v", testName, err)
 		return fmt.Errorf("%v: %w", testName, err)
 	}
 
@@ -1634,7 +1634,7 @@ func CreateBucket_non_default_acl(s *S3Conf) error {
 
 	err = teardown(s, bucket)
 	if err != nil {
-		failF("%v: %v", err)
+		failF("%v: %v", testName, err)
 		return fmt.Errorf("%v: %w", testName, err)
 	}
 
@@ -1658,7 +1658,7 @@ func CreateBucket_default_object_lock(s *S3Conf) error {
 	})
 	cancel()
 	if err != nil {
-		failF("%v: %v", err)
+		failF("%v: %v", testName, err)
 		return fmt.Errorf("%v: %w", testName, err)
 	}
 
@@ -1668,7 +1668,7 @@ func CreateBucket_default_object_lock(s *S3Conf) error {
 	})
 	cancel()
 	if err != nil {
-		failF("%v: %v", err)
+		failF("%v: %v", testName, err)
 		return fmt.Errorf("%v: %w", testName, err)
 	}
 
@@ -1679,7 +1679,7 @@ func CreateBucket_default_object_lock(s *S3Conf) error {
 
 	err = teardown(s, bucket)
 	if err != nil {
-		failF("%v: %v", err)
+		failF("%v: %v", testName, err)
 		return fmt.Errorf("%v: %w", testName, err)
 	}
 
@@ -1909,7 +1909,7 @@ func ListBuckets_as_admin(s *S3Conf) error {
 		}
 		if !compareBuckets(out.Buckets, buckets) {
 			return fmt.Errorf("expected list buckets result to be %v, instead got %v",
-				buckets, out.Buckets)
+				sprintBuckets(buckets), sprintBuckets(out.Buckets))
 		}
 
 		for _, elem := range buckets[1:] {
@@ -2047,7 +2047,7 @@ func ListBuckets_truncated(s *S3Conf) error {
 		}
 		if !compareBuckets(out.Buckets, buckets[:maxBuckets]) {
 			return fmt.Errorf("expected list buckets result to be %v, instead got %v",
-				buckets[:maxBuckets], out.Buckets)
+				sprintBuckets(buckets[:maxBuckets]), sprintBuckets(out.Buckets))
 		}
 		if getString(out.ContinuationToken) != getString(buckets[maxBuckets-1].Name) {
 			return fmt.Errorf("expected ContinuationToken to be %v, instead got %v",
@@ -2065,7 +2065,7 @@ func ListBuckets_truncated(s *S3Conf) error {
 
 		if !compareBuckets(out.Buckets, buckets[maxBuckets:]) {
 			return fmt.Errorf("expected list buckets result to be %v, instead got %v",
-				buckets[maxBuckets:], out.Buckets)
+				sprintBuckets(buckets[:maxBuckets]), sprintBuckets(out.Buckets))
 		}
 		if out.ContinuationToken != nil {
 			return fmt.Errorf("expected nil continuation token, instead got %v",
@@ -2098,7 +2098,7 @@ func ListBuckets_empty_success(s *S3Conf) error {
 
 		if len(out.Buckets) > 0 {
 			return fmt.Errorf("expected list buckets result to be %v, instead got %v",
-				[]types.Bucket{}, out.Buckets)
+				[]types.Bucket{}, sprintBuckets(out.Buckets))
 		}
 
 		return nil
@@ -2136,7 +2136,7 @@ func ListBuckets_success(s *S3Conf) error {
 		}
 		if !compareBuckets(out.Buckets, buckets) {
 			return fmt.Errorf("expected list buckets result to be %v, instead got %v",
-				buckets, out.Buckets)
+				sprintBuckets(buckets), sprintBuckets(out.Buckets))
 		}
 
 		for _, elem := range buckets[1:] {
@@ -2157,14 +2157,14 @@ func CreateDeleteBucket_success(s *S3Conf) error {
 
 	err := setup(s, bucket)
 	if err != nil {
-		failF("%v: %v", err)
+		failF("%v: %v", testName, err)
 		return fmt.Errorf("%v: %w", testName, err)
 
 	}
 
 	err = teardown(s, bucket)
 	if err != nil {
-		failF("%v: %v", err)
+		failF("%v: %v", testName, err)
 		return fmt.Errorf("%v: %w", testName, err)
 	}
 
@@ -3018,13 +3018,13 @@ func PutObject_with_object_lock(s *S3Conf) error {
 	}
 
 	if err := changeBucketObjectLockStatus(client, bucket, false); err != nil {
-		failF("%v: %v", err)
+		failF("%v: %v", testName, err)
 		return fmt.Errorf("%v: %w", testName, err)
 	}
 
 	err = teardown(s, bucket)
 	if err != nil {
-		failF("%v: %v", err)
+		failF("%v: %v", testName, err)
 		return fmt.Errorf("%v: %w", testName, err)
 	}
 
@@ -3480,7 +3480,7 @@ func PutObject_racey_success(s *S3Conf) error {
 
 	err = teardown(s, bucket)
 	if err != nil {
-		failF("%v: %v", err)
+		failF("%v: %v", testName, err)
 		return fmt.Errorf("%v: %w", testName, err)
 	}
 
@@ -6161,7 +6161,7 @@ func ListObjects_non_truncated_common_prefixes(s *S3Conf) error {
 		cPrefs := []string{"cquux/"}
 		if !comparePrefixes(cPrefs, res.CommonPrefixes) {
 			return fmt.Errorf("expected common prefixes to be %v, instead got %+v",
-				cPrefs, res.CommonPrefixes)
+				cPrefs, sprintPrefixes(res.CommonPrefixes))
 		}
 
 		return nil
@@ -6438,7 +6438,8 @@ func ListObjectsV2_truncated_common_prefixes(s *S3Conf) error {
 		}
 
 		if !comparePrefixes([]string{"d1/", "d2/", "d3/"}, out.CommonPrefixes) {
-			return fmt.Errorf("expected the common prefixes to be %v, instead got %v", []string{"d1/", "d2/", "d3/"}, out.CommonPrefixes)
+			return fmt.Errorf("expected the common prefixes to be %v, instead got %v",
+				[]string{"d1/", "d2/", "d3/"}, sprintPrefixes(out.CommonPrefixes))
 		}
 
 		if out.MaxKeys == nil {
@@ -6466,7 +6467,7 @@ func ListObjectsV2_truncated_common_prefixes(s *S3Conf) error {
 
 		if !comparePrefixes([]string{"d4/"}, out.CommonPrefixes) {
 			return fmt.Errorf("expected the common prefixes to be %v, instead got %v",
-				[]string{"d4/"}, out.CommonPrefixes)
+				[]string{"d4/"}, sprintPrefixes(out.CommonPrefixes))
 		}
 		if getString(out.Delimiter) != delim {
 			return fmt.Errorf("expected the delimiter to be %v, instead got %v",
@@ -6523,7 +6524,7 @@ func ListObjectsV2_non_truncated_common_prefixes(s *S3Conf) error {
 		cPrefs := []string{"cquux/"}
 		if !comparePrefixes(cPrefs, res.CommonPrefixes) {
 			return fmt.Errorf("expected common prefixes to be %v, instead got %+v",
-				cPrefs, res.CommonPrefixes)
+				cPrefs, sprintPrefixes(res.CommonPrefixes))
 		}
 
 		return nil

tests/integration/utils.go

@@ -1813,3 +1813,29 @@ func calculateEtag(data []byte) (string, error) {
 	dataSum := h.Sum(nil)
 	return fmt.Sprintf("\"%s\"", hex.EncodeToString(dataSum[:])), nil
 }
+
+func sprintBuckets(buckets []types.Bucket) string {
+	if len(buckets) == 0 {
+		return ""
+	}
+
+	names := make([]string, len(buckets))
+	for i, bucket := range buckets {
+		names[i] = *bucket.Name
+	}
+
+	return strings.Join(names, ",")
+}
+
+func sprintPrefixes(cpfx []types.CommonPrefix) string {
+	if len(cpfx) == 0 {
+		return ""
+	}
+
+	names := make([]string, len(cpfx))
+	for i, pfx := range cpfx {
+		names[i] = *pfx.Prefix
+	}
+
+	return strings.Join(names, ",")
+}