satellite/metabase: ListObjects

Current metainfo.ListObjects implementation is using metabase iterator to list objects.
In the non-recursive case, it used to retrieve all the corresponding rows and then discarded the entries that did not fit the listing request.
This can lead in some edge cases (each prefix contains more than batchsize objects/sub-prefixes) to make unecessary calls to the db.

This change defines the metabase.ListObjects and aims at retrieving only prefixes (but not objects under it) and objects by modifying the SQL query.

In this version, it is not optimized on the database side. Cockroach will still have to go through all rows under a prefix, so there is still room for improvement.

metainfo.ListObjects is not currently using this method as we would like to assess its performance on the QA satellite first.

Fixes https://github.com/storj/storj/issues/5088

Change-Id: Ied3a9210871871d9d4a3096888d3e40c2dceed61

satellite/metabase/list_objects.go

@@ -0,0 +1,250 @@
+// Copyright (C) 2022 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package metabase
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"strings"
+
+	"storj.io/common/uuid"
+	"storj.io/private/tagsql"
+)
+
+// ListObjectsCursor is a cursor used during iteration through objects.
+type ListObjectsCursor struct {
+	Key     ObjectKey
+	Version Version
+}
+
+// ListObjects contains arguments necessary for listing objects.
+type ListObjects struct {
+	ProjectID             uuid.UUID
+	BucketName            string
+	Recursive             bool
+	Limit                 int
+	Prefix                ObjectKey
+	Cursor                ListObjectsCursor
+	Status                ObjectStatus
+	IncludeCustomMetadata bool
+	IncludeSystemMetadata bool
+}
+
+// Verify verifies get object request fields.
+func (opts *ListObjects) Verify() error {
+	switch {
+	case opts.ProjectID.IsZero():
+		return ErrInvalidRequest.New("ProjectID missing")
+	case opts.BucketName == "":
+		return ErrInvalidRequest.New("BucketName missing")
+	case opts.Limit < 0:
+		return ErrInvalidRequest.New("Invalid limit: %d", opts.Limit)
+	case !(opts.Status == Pending || opts.Status == Committed):
+		return ErrInvalidRequest.New("Status is invalid")
+	}
+	return nil
+}
+
+// ListObjectsResult result of listing objects.
+type ListObjectsResult struct {
+	Objects []ObjectEntry
+	More    bool
+}
+
+// ListObjects lists objects.
+func (db *DB) ListObjects(ctx context.Context, opts ListObjects) (result ListObjectsResult, err error) {
+	defer mon.Task()(&ctx)(&err)
+
+	if err := opts.Verify(); err != nil {
+		return ListObjectsResult{}, err
+	}
+
+	ListLimit.Ensure(&opts.Limit)
+
+	var entries []ObjectEntry
+	err = withRows(db.db.QueryContext(ctx, opts.getSQLQuery(),
+		opts.ProjectID, opts.BucketName, opts.startKey(), opts.Cursor.Version,
+		opts.stopKey(), opts.Status,
+		opts.Limit+1, len(opts.Prefix)+1))(func(rows tagsql.Rows) error {
+		entries, err = scanListObjectsResult(rows, opts)
+		return err
+	})
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return ListObjectsResult{}, nil
+		}
+		return ListObjectsResult{}, Error.New("unable to list objects: %w", err)
+	}
+
+	if len(entries) > opts.Limit {
+		result.More = true
+		result.Objects = entries[:opts.Limit]
+		return result, nil
+	}
+
+	result.Objects = entries
+	result.More = false
+	return result, nil
+}
+
+func (opts *ListObjects) getSQLQuery() string {
+	return `
+	SELECT ` + opts.selectedFields() + `
+	FROM objects
+	WHERE
+		(project_id, bucket_name, object_key, version) > ($1, $2, $3, $4)
+		AND ` + opts.stopCondition() + `
+		AND status = $6
+		AND (expires_at IS NULL OR expires_at > now())
+	ORDER BY ` + opts.orderBy() + `
+	LIMIT $7
+	`
+}
+
+func (opts *ListObjects) stopKey() []byte {
+	if opts.Prefix != "" {
+		return []byte(prefixLimit(opts.Prefix))
+	}
+	return nextBucket([]byte(opts.BucketName))
+}
+
+func (opts *ListObjects) stopCondition() string {
+	if opts.Prefix != "" {
+		return "(project_id, bucket_name, object_key) < ($1, $2, $5)"
+	}
+	return "(project_id, bucket_name) < ($1, $5)"
+}
+
+func (opts *ListObjects) orderBy() string {
+	if !opts.Recursive {
+		return "entry_key ASC"
+	}
+
+	return "(object_key, version) ASC"
+}
+
+func (opts ListObjects) selectedFields() (selectedFields string) {
+
+	if opts.Recursive {
+		selectedFields = `
+			substring(object_key from $8), FALSE as is_prefix`
+	} else {
+		selectedFields = `
+			DISTINCT ON (entry_key)
+			CASE
+				WHEN position('/' IN substring(object_key from $8)) <> 0
+				THEN substring(substring(object_key from $8) from 0 for (position('/' IN substring(object_key from $8)) +1))
+				ELSE substring(object_key from $8)
+			END
+			AS entry_key,
+			position('/' IN substring(object_key from $8)) <> 0 AS is_prefix`
+	}
+
+	selectedFields += `
+	,stream_id
+	,version
+	,encryption`
+
+	if opts.IncludeSystemMetadata {
+		selectedFields += `
+		,status
+		,created_at
+		,expires_at
+		,segment_count
+		,total_plain_size
+		,total_encrypted_size
+		,fixed_segment_size`
+	}
+
+	if opts.IncludeCustomMetadata {
+		selectedFields += `
+		,encrypted_metadata_nonce
+		,encrypted_metadata
+		,encrypted_metadata_encrypted_key`
+	}
+	return selectedFields
+}
+
+// startKey determines what should be the starting key for the given options.
+// in the recursive case, or if the cursor key is not in the specified prefix,
+// we start at the greatest key between cursor and prefix.
+// Otherwise (non-recursive), we start at the prefix after the one in the cursor.
+func (opts *ListObjects) startKey() ObjectKey {
+	if opts.Prefix == "" && opts.Cursor.Key == "" {
+		return ""
+	}
+	if opts.Recursive || !strings.HasPrefix(string(opts.Cursor.Key), string(opts.Prefix)) {
+		if lessKey(opts.Cursor.Key, opts.Prefix) {
+			return opts.Prefix
+		}
+		return opts.Cursor.Key
+	}
+
+	// in the recursive case
+	// prefix | cursor | startKey
+	// a/b/   | a/b/c/d/e | c/d/[0xff] (the first prefix/object key we return )
+	key := opts.Cursor.Key
+	prefixSize := len(opts.Prefix)
+	subPrefix := key[prefixSize:] // c/d/e
+
+	firstDelimiter := strings.Index(string(subPrefix), string(Delimiter))
+	if firstDelimiter == -1 {
+		return key
+	}
+	newKey := []byte(key[:prefixSize+firstDelimiter+1]) // c/d/
+	newKey = append(newKey, 0xff)
+	return ObjectKey(newKey)
+}
+
+func scanListObjectsResult(rows tagsql.Rows, opts ListObjects) (entries []ObjectEntry, err error) {
+
+	for rows.Next() {
+		var item ObjectEntry
+
+		fields := []interface{}{
+			&item.ObjectKey,
+			&item.IsPrefix,
+			&item.StreamID,
+			&item.Version,
+			encryptionParameters{&item.Encryption},
+		}
+
+		if opts.IncludeSystemMetadata {
+			fields = append(fields,
+				&item.Status,
+				&item.CreatedAt,
+				&item.ExpiresAt,
+				&item.SegmentCount,
+				&item.TotalPlainSize,
+				&item.TotalEncryptedSize,
+				&item.FixedSegmentSize,
+			)
+		}
+
+		if opts.IncludeCustomMetadata {
+			fields = append(fields,
+				&item.EncryptedMetadataNonce,
+				&item.EncryptedMetadata,
+				&item.EncryptedMetadataEncryptedKey,
+			)
+		}
+
+		if err := rows.Scan(fields...); err != nil {
+			return entries, err
+		}
+
+		if item.IsPrefix {
+			item = ObjectEntry{
+				IsPrefix:  true,
+				ObjectKey: item.ObjectKey,
+				Status:    opts.Status,
+			}
+		}
+
+		entries = append(entries, item)
+	}
+
+	return entries, nil
+}

satellite/metabase/list_objects_test.go

@@ -0,0 +1,900 @@
+// Copyright (C) 2022 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package metabase_test
+
+import (
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"storj.io/common/testcontext"
+	"storj.io/common/uuid"
+	"storj.io/storj/satellite/metabase"
+	"storj.io/storj/satellite/metabase/metabasetest"
+)
+
+func TestListObjects(t *testing.T) {
+	metabasetest.Run(t, func(ctx *testcontext.Context, t *testing.T, db *metabase.DB) {
+		obj := metabasetest.RandObjectStream()
+
+		t.Run("ProjectID missing", func(t *testing.T) {
+			defer metabasetest.DeleteAll{}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts:     metabase.ListObjects{},
+				ErrClass: &metabase.ErrInvalidRequest,
+				ErrText:  "ProjectID missing",
+			}.Check(ctx, t, db)
+
+			metabasetest.Verify{}.Check(ctx, t, db)
+		})
+
+		t.Run("BucketName missing", func(t *testing.T) {
+			defer metabasetest.DeleteAll{}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID: obj.ProjectID,
+					Limit:     -1,
+				},
+				ErrClass: &metabase.ErrInvalidRequest,
+				ErrText:  "BucketName missing",
+			}.Check(ctx, t, db)
+
+			metabasetest.Verify{}.Check(ctx, t, db)
+		})
+
+		t.Run("Invalid limit", func(t *testing.T) {
+			defer metabasetest.DeleteAll{}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:  obj.ProjectID,
+					BucketName: obj.BucketName,
+					Limit:      -1,
+				},
+				ErrClass: &metabase.ErrInvalidRequest,
+				ErrText:  "Invalid limit: -1",
+			}.Check(ctx, t, db)
+
+			metabasetest.Verify{}.Check(ctx, t, db)
+		})
+
+		t.Run("Status is invalid", func(t *testing.T) {
+			defer metabasetest.DeleteAll{}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:  obj.ProjectID,
+					BucketName: obj.BucketName,
+					Limit:      3,
+				},
+				ErrClass: &metabase.ErrInvalidRequest,
+				ErrText:  "Status is invalid",
+			}.Check(ctx, t, db)
+
+			metabasetest.Verify{}.Check(ctx, t, db)
+		})
+
+		t.Run("no objects", func(t *testing.T) {
+			defer metabasetest.DeleteAll{}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:  obj.ProjectID,
+					BucketName: obj.BucketName,
+					Status:     metabase.Committed,
+				},
+				Result: metabase.ListObjectsResult{},
+			}.Check(ctx, t, db)
+
+			metabasetest.Verify{}.Check(ctx, t, db)
+		})
+
+		t.Run("less objects than limit", func(t *testing.T) {
+			defer metabasetest.DeleteAll{}.Check(ctx, t, db)
+			numberOfObjects := 3
+			limit := 10
+			expected := make([]metabase.ObjectEntry, numberOfObjects)
+			objects := createObjects(ctx, t, db, numberOfObjects, uuid.UUID{1}, "mybucket")
+			for i, obj := range objects {
+				if delimiterIndex := strings.Index(string(obj.ObjectKey), string(metabase.Delimiter)); delimiterIndex > -1 {
+					expected[i] = metabase.ObjectEntry{
+						IsPrefix:  true,
+						ObjectKey: obj.ObjectKey[:delimiterIndex+1],
+						Status:    3,
+					}
+				} else {
+					expected[i] = objectEntryFromRaw(obj)
+				}
+			}
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             uuid.UUID{1},
+					BucketName:            "mybucket",
+					Recursive:             false,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+					Limit:                 limit,
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: expected,
+					More:    false,
+				}}.Check(ctx, t, db)
+			metabasetest.Verify{Objects: objects}.Check(ctx, t, db)
+		})
+
+		t.Run("more objects than limit", func(t *testing.T) {
+			defer metabasetest.DeleteAll{}.Check(ctx, t, db)
+			numberOfObjects := 10
+			limit := 3
+			expected := make([]metabase.ObjectEntry, limit)
+			objects := createObjects(ctx, t, db, numberOfObjects, uuid.UUID{1}, "mybucket")
+			for i, obj := range objects[:limit] {
+				expected[i] = objectEntryFromRaw(obj)
+			}
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             uuid.UUID{1},
+					BucketName:            "mybucket",
+					Recursive:             true,
+					Limit:                 limit,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: expected,
+					More:    true,
+				}}.Check(ctx, t, db)
+			metabasetest.Verify{Objects: objects}.Check(ctx, t, db)
+		})
+
+		t.Run("objects in one bucket in project with 2 buckets", func(t *testing.T) {
+			defer metabasetest.DeleteAll{}.Check(ctx, t, db)
+			numberOfObjectsPerBucket := 5
+
+			expected := make([]metabase.ObjectEntry, numberOfObjectsPerBucket)
+
+			objectsBucketA := createObjects(ctx, t, db, numberOfObjectsPerBucket, uuid.UUID{1}, "bucket-a")
+			objectsBucketB := createObjects(ctx, t, db, numberOfObjectsPerBucket, uuid.UUID{1}, "bucket-b")
+
+			for i, obj := range objectsBucketA {
+				expected[i] = objectEntryFromRaw(obj)
+			}
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             uuid.UUID{1},
+					BucketName:            "bucket-a",
+					Recursive:             true,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: expected,
+				}}.Check(ctx, t, db)
+
+			metabasetest.Verify{
+				Objects: append(objectsBucketA, objectsBucketB...),
+			}.Check(ctx, t, db)
+		})
+
+		t.Run("objects in one bucket with same bucketName in another project", func(t *testing.T) {
+			defer metabasetest.DeleteAll{}.Check(ctx, t, db)
+			numberOfObjectsPerBucket := 5
+
+			expected := make([]metabase.ObjectEntry, numberOfObjectsPerBucket)
+
+			objectsProject1 := createObjects(ctx, t, db, numberOfObjectsPerBucket, uuid.UUID{1}, "mybucket")
+			objectsProject2 := createObjects(ctx, t, db, numberOfObjectsPerBucket, uuid.UUID{2}, "mybucket")
+			for i, obj := range objectsProject1 {
+				expected[i] = objectEntryFromRaw(obj)
+			}
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             uuid.UUID{1},
+					BucketName:            "mybucket",
+					Recursive:             true,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: expected,
+				}}.Check(ctx, t, db)
+
+			metabasetest.Verify{
+				Objects: append(objectsProject1, objectsProject2...),
+			}.Check(ctx, t, db)
+		})
+
+		t.Run("recursive", func(t *testing.T) {
+			defer metabasetest.DeleteAll{}.Check(ctx, t, db)
+			projectID, bucketName := uuid.UUID{1}, "bucky"
+
+			objects := createObjectsWithKeys(ctx, t, db, projectID, bucketName, []metabase.ObjectKey{
+				"a",
+				"b/1",
+				"b/2",
+				"b/3",
+				"c",
+				"c/",
+				"c//",
+				"c/1",
+				"g",
+			})
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             projectID,
+					BucketName:            bucketName,
+					Recursive:             true,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: []metabase.ObjectEntry{
+						objects["a"],
+						objects["b/1"],
+						objects["b/2"],
+						objects["b/3"],
+						objects["c"],
+						objects["c/"],
+						objects["c//"],
+						objects["c/1"],
+						objects["g"],
+					},
+				}}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             projectID,
+					BucketName:            bucketName,
+					Recursive:             true,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+
+					Cursor: metabase.ListObjectsCursor{Key: "a", Version: 10},
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: []metabase.ObjectEntry{
+						objects["b/1"],
+						objects["b/2"],
+						objects["b/3"],
+						objects["c"],
+						objects["c/"],
+						objects["c//"],
+						objects["c/1"],
+						objects["g"],
+					}},
+			}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             projectID,
+					BucketName:            bucketName,
+					Recursive:             true,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+
+					Cursor: metabase.ListObjectsCursor{Key: "b", Version: 0},
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: []metabase.ObjectEntry{
+						objects["b/1"],
+						objects["b/2"],
+						objects["b/3"],
+						objects["c"],
+						objects["c/"],
+						objects["c//"],
+						objects["c/1"],
+						objects["g"],
+					},
+				}}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             projectID,
+					BucketName:            bucketName,
+					Recursive:             true,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+
+					Prefix: "b/",
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: withoutPrefix("b/",
+						objects["b/1"],
+						objects["b/2"],
+						objects["b/3"],
+					),
+				}}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             projectID,
+					BucketName:            bucketName,
+					Recursive:             true,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+
+					Prefix: "b/",
+					Cursor: metabase.ListObjectsCursor{Key: "a"},
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: withoutPrefix("b/",
+						objects["b/1"],
+						objects["b/2"],
+						objects["b/3"],
+					),
+				}}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             projectID,
+					BucketName:            bucketName,
+					Recursive:             true,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+
+					Prefix: "b/",
+					Cursor: metabase.ListObjectsCursor{Key: "b/2", Version: -3},
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: withoutPrefix("b/",
+						objects["b/2"],
+						objects["b/3"],
+					),
+				}}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             projectID,
+					BucketName:            bucketName,
+					Recursive:             true,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+
+					Prefix: "b/",
+					Cursor: metabase.ListObjectsCursor{Key: "c/"},
+				},
+				Result: metabase.ListObjectsResult{},
+			}.Check(ctx, t, db)
+		})
+
+		t.Run("non-recursive", func(t *testing.T) {
+			defer metabasetest.DeleteAll{}.Check(ctx, t, db)
+			projectID, bucketName := uuid.UUID{1}, "bucky"
+
+			objects := createObjectsWithKeys(ctx, t, db, projectID, bucketName, []metabase.ObjectKey{
+				"a",
+				"b/1",
+				"b/2",
+				"b/3",
+				"c",
+				"c/",
+				"c//",
+				"c/1",
+				"g",
+			})
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             projectID,
+					BucketName:            bucketName,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: []metabase.ObjectEntry{
+						objects["a"],
+						prefixEntry("b/", metabase.Committed),
+						objects["c"],
+						prefixEntry("c/", metabase.Committed),
+						objects["g"],
+					}},
+			}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             projectID,
+					BucketName:            bucketName,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+
+					Cursor: metabase.ListObjectsCursor{Key: "a", Version: 10},
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: []metabase.ObjectEntry{
+						prefixEntry("b/", metabase.Committed),
+						objects["c"],
+						prefixEntry("c/", metabase.Committed),
+						objects["g"],
+					}},
+			}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             projectID,
+					BucketName:            bucketName,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+
+					Cursor: metabase.ListObjectsCursor{Key: "b", Version: 0},
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: []metabase.ObjectEntry{
+						prefixEntry("b/", metabase.Committed),
+						objects["c"],
+						prefixEntry("c/", metabase.Committed),
+						objects["g"],
+					}},
+			}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             projectID,
+					BucketName:            bucketName,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+
+					Prefix: "b/",
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: withoutPrefix("b/",
+						objects["b/1"],
+						objects["b/2"],
+						objects["b/3"],
+					)},
+			}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             projectID,
+					BucketName:            bucketName,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+
+					Prefix: "b/",
+					Cursor: metabase.ListObjectsCursor{Key: "a"},
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: withoutPrefix("b/",
+						objects["b/1"],
+						objects["b/2"],
+						objects["b/3"],
+					)},
+			}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             projectID,
+					BucketName:            bucketName,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+
+					Prefix: "b/",
+					Cursor: metabase.ListObjectsCursor{Key: "b/2", Version: -3},
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: withoutPrefix("b/",
+						objects["b/2"],
+						objects["b/3"],
+					)},
+			}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             projectID,
+					BucketName:            bucketName,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+
+					Prefix: "b/",
+					Cursor: metabase.ListObjectsCursor{Key: "c/"},
+				},
+				Result: metabase.ListObjectsResult{},
+			}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             projectID,
+					BucketName:            bucketName,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+
+					Prefix: "c/",
+					Cursor: metabase.ListObjectsCursor{Key: "c/"},
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: withoutPrefix("c/",
+						objects["c/"],
+						prefixEntry("c//", metabase.Committed),
+						objects["c/1"],
+					)},
+			}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:             projectID,
+					BucketName:            bucketName,
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+
+					Prefix: "c//",
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: withoutPrefix("c//",
+						objects["c//"],
+					)},
+			}.Check(ctx, t, db)
+		})
+
+	})
+}
+
+func TestListObjectsSkipCursor(t *testing.T) {
+	metabasetest.Run(t, func(ctx *testcontext.Context, t *testing.T, db *metabase.DB) {
+		projectID, bucketName := uuid.UUID{1}, "bucky"
+
+		t.Run("no prefix", func(t *testing.T) {
+			defer metabasetest.DeleteAll{}.Check(ctx, t, db)
+
+			createObjectsWithKeys(ctx, t, db, projectID, bucketName, []metabase.ObjectKey{
+				"08/test",
+				"09/test",
+				"10/test",
+			})
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:  projectID,
+					BucketName: bucketName,
+					Recursive:  false,
+					Prefix:     "",
+					Cursor: metabase.ListObjectsCursor{
+						Key:     metabase.ObjectKey("08/"),
+						Version: 1,
+					},
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: []metabase.ObjectEntry{
+						prefixEntry(metabase.ObjectKey("09/"), metabase.Committed),
+						prefixEntry(metabase.ObjectKey("10/"), metabase.Committed),
+					}},
+			}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:  projectID,
+					BucketName: bucketName,
+					Recursive:  false,
+					Prefix:     "",
+					Cursor: metabase.ListObjectsCursor{
+						Key:     metabase.ObjectKey("08"),
+						Version: 1,
+					},
+					Status:                metabase.Committed,
+					IncludeSystemMetadata: true,
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: []metabase.ObjectEntry{
+						prefixEntry(metabase.ObjectKey("08/"), metabase.Committed),
+						prefixEntry(metabase.ObjectKey("09/"), metabase.Committed),
+						prefixEntry(metabase.ObjectKey("10/"), metabase.Committed),
+					}},
+			}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:  projectID,
+					BucketName: bucketName,
+					Recursive:  false,
+					Prefix:     "",
+					Cursor: metabase.ListObjectsCursor{
+						Key:     metabase.ObjectKey("08/a/x"),
+						Version: 1,
+					},
+					Status:                metabase.Committed,
+					IncludeSystemMetadata: true,
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: []metabase.ObjectEntry{
+						prefixEntry(metabase.ObjectKey("09/"), metabase.Committed),
+						prefixEntry(metabase.ObjectKey("10/"), metabase.Committed),
+					}},
+			}.Check(ctx, t, db)
+		})
+
+		t.Run("prefix", func(t *testing.T) {
+			defer metabasetest.DeleteAll{}.Check(ctx, t, db)
+
+			createObjectsWithKeys(ctx, t, db, projectID, bucketName, []metabase.ObjectKey{
+				"2017/05/08/test",
+				"2017/05/09/test",
+				"2017/05/10/test",
+			})
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:  projectID,
+					BucketName: bucketName,
+					Recursive:  false,
+					Prefix:     metabase.ObjectKey("2017/05/"),
+					Cursor: metabase.ListObjectsCursor{
+						Key:     metabase.ObjectKey("2017/05/08"),
+						Version: 1,
+					},
+					Status:                metabase.Committed,
+					IncludeSystemMetadata: true,
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: []metabase.ObjectEntry{
+						prefixEntry(metabase.ObjectKey("08/"), metabase.Committed),
+						prefixEntry(metabase.ObjectKey("09/"), metabase.Committed),
+						prefixEntry(metabase.ObjectKey("10/"), metabase.Committed),
+					}},
+			}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:  projectID,
+					BucketName: bucketName,
+					Recursive:  false,
+					Prefix:     metabase.ObjectKey("2017/05/"),
+					Cursor: metabase.ListObjectsCursor{
+						Key:     metabase.ObjectKey("2017/05/08/"),
+						Version: 1,
+					},
+					Status:                metabase.Committed,
+					IncludeSystemMetadata: true,
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: []metabase.ObjectEntry{
+						prefixEntry(metabase.ObjectKey("09/"), metabase.Committed),
+						prefixEntry(metabase.ObjectKey("10/"), metabase.Committed),
+					}},
+			}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:  projectID,
+					BucketName: bucketName,
+					Recursive:  false,
+					Prefix:     metabase.ObjectKey("2017/05/"),
+					Cursor: metabase.ListObjectsCursor{
+						Key:     metabase.ObjectKey("2017/05/08/a/x"),
+						Version: 1,
+					},
+					Status:                metabase.Committed,
+					IncludeSystemMetadata: true,
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: []metabase.ObjectEntry{
+						prefixEntry(metabase.ObjectKey("09/"), metabase.Committed),
+						prefixEntry(metabase.ObjectKey("10/"), metabase.Committed),
+					}},
+			}.Check(ctx, t, db)
+		})
+
+		t.Run("batch-size", func(t *testing.T) {
+			defer metabasetest.DeleteAll{}.Check(ctx, t, db)
+
+			afterDelimiter := metabase.ObjectKey(metabase.Delimiter + 1)
+
+			objects := createObjectsWithKeys(ctx, t, db, projectID, bucketName, []metabase.ObjectKey{
+				"2017/05/08",
+				"2017/05/08/a",
+				"2017/05/08/b",
+				"2017/05/08/c",
+				"2017/05/08/d",
+				"2017/05/08/e",
+				"2017/05/08" + afterDelimiter,
+				"2017/05/09/a",
+				"2017/05/09/b",
+				"2017/05/09/c",
+				"2017/05/09/d",
+				"2017/05/09/e",
+				"2017/05/10/a",
+				"2017/05/10/b",
+				"2017/05/10/c",
+				"2017/05/10/d",
+				"2017/05/10/e",
+			})
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:  projectID,
+					BucketName: bucketName,
+					Recursive:  false,
+					Prefix:     metabase.ObjectKey("2017/05/"),
+					Cursor: metabase.ListObjectsCursor{
+						Key:     metabase.ObjectKey("2017/05/08"),
+						Version: 1,
+					},
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: []metabase.ObjectEntry{
+						prefixEntry(metabase.ObjectKey("08/"), metabase.Committed),
+						withoutPrefix1("2017/05/", objects["2017/05/08"+afterDelimiter]),
+						prefixEntry(metabase.ObjectKey("09/"), metabase.Committed),
+						prefixEntry(metabase.ObjectKey("10/"), metabase.Committed),
+					}},
+			}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:  projectID,
+					BucketName: bucketName,
+					Recursive:  false,
+					//BatchSize:  3,
+					Prefix: metabase.ObjectKey("2017/05/"),
+					Cursor: metabase.ListObjectsCursor{
+						Key:     metabase.ObjectKey("2017/05/08/"),
+						Version: 1,
+					},
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: []metabase.ObjectEntry{
+						withoutPrefix1("2017/05/", objects["2017/05/08"+afterDelimiter]),
+						prefixEntry(metabase.ObjectKey("09/"), metabase.Committed),
+						prefixEntry(metabase.ObjectKey("10/"), metabase.Committed),
+					}},
+			}.Check(ctx, t, db)
+
+			metabasetest.ListObjects{
+				Opts: metabase.ListObjects{
+					ProjectID:  projectID,
+					BucketName: bucketName,
+					Recursive:  false,
+					Prefix:     metabase.ObjectKey("2017/05/"),
+					Cursor: metabase.ListObjectsCursor{
+						Key:     metabase.ObjectKey("2017/05/08/a/x"),
+						Version: 1,
+					},
+					Status:                metabase.Committed,
+					IncludeCustomMetadata: true,
+					IncludeSystemMetadata: true,
+				},
+				Result: metabase.ListObjectsResult{
+					Objects: []metabase.ObjectEntry{
+						withoutPrefix1("2017/05/", objects["2017/05/08"+afterDelimiter]),
+						prefixEntry(metabase.ObjectKey("09/"), metabase.Committed),
+						prefixEntry(metabase.ObjectKey("10/"), metabase.Committed),
+					}},
+			}.Check(ctx, t, db)
+		})
+	})
+}
+
+func BenchmarkNonRecursiveObjectsListing(b *testing.B) {
+	metabasetest.Bench(b, func(ctx *testcontext.Context, b *testing.B, db *metabase.DB) {
+		baseObj := metabasetest.RandObjectStream()
+
+		batchsize := 5
+		for i := 0; i < 500; i++ {
+			metabasetest.CreateObject(ctx, b, db, metabasetest.RandObjectStream(), 0)
+		}
+
+		for i := 0; i < 10; i++ {
+			baseObj.ObjectKey = metabase.ObjectKey("foo/" + strconv.Itoa(i))
+			metabasetest.CreateObject(ctx, b, db, baseObj, 0)
+
+			baseObj.ObjectKey = metabase.ObjectKey("foo/prefixA/" + strconv.Itoa(i))
+			metabasetest.CreateObject(ctx, b, db, baseObj, 0)
+
+			baseObj.ObjectKey = metabase.ObjectKey("foo/prefixB/" + strconv.Itoa(i))
+			metabasetest.CreateObject(ctx, b, db, baseObj, 0)
+		}
+
+		for i := 0; i < 50; i++ {
+			baseObj.ObjectKey = metabase.ObjectKey("boo/foo" + strconv.Itoa(i) + "/object")
+			metabasetest.CreateObject(ctx, b, db, baseObj, 0)
+		}
+
+		b.Run("listing no prefix", func(b *testing.B) {
+			for i := 0; i < b.N; i++ {
+				result, err := db.ListObjects(ctx, metabase.ListObjects{
+					ProjectID:  baseObj.ProjectID,
+					BucketName: baseObj.BucketName,
+					Status:     metabase.Committed,
+					Limit:      batchsize,
+				})
+				require.NoError(b, err)
+				for result.More {
+					result, err = db.ListObjects(ctx, metabase.ListObjects{
+						ProjectID:  baseObj.ProjectID,
+						BucketName: baseObj.BucketName,
+						Cursor:     metabase.ListObjectsCursor{Key: result.Objects[len(result.Objects)-1].ObjectKey},
+						Status:     metabase.Committed,
+						Limit:      batchsize,
+					})
+					require.NoError(b, err)
+				}
+			}
+		})
+
+		b.Run("listing with prefix", func(b *testing.B) {
+			for i := 0; i < b.N; i++ {
+				result, err := db.ListObjects(ctx, metabase.ListObjects{
+					ProjectID:  baseObj.ProjectID,
+					BucketName: baseObj.BucketName,
+					Prefix:     "foo/",
+					Status:     metabase.Committed,
+					Limit:      batchsize,
+				})
+				require.NoError(b, err)
+				for result.More {
+					cursorKey := "foo/" + result.Objects[len(result.Objects)-1].ObjectKey
+					result, err = db.ListObjects(ctx, metabase.ListObjects{
+						ProjectID:  baseObj.ProjectID,
+						BucketName: baseObj.BucketName,
+						Prefix:     "foo/",
+						Cursor:     metabase.ListObjectsCursor{Key: cursorKey},
+						Status:     metabase.Committed,
+						Limit:      batchsize,
+					})
+					require.NoError(b, err)
+				}
+			}
+		})
+
+		b.Run("listing only prefix", func(b *testing.B) {
+			for i := 0; i < b.N; i++ {
+				result, err := db.ListObjects(ctx, metabase.ListObjects{
+					ProjectID:  baseObj.ProjectID,
+					BucketName: baseObj.BucketName,
+					Prefix:     "boo/",
+					Status:     metabase.Committed,
+					Limit:      batchsize,
+				})
+				require.NoError(b, err)
+				for result.More {
+					cursorKey := "boo/" + result.Objects[len(result.Objects)-1].ObjectKey
+					result, err = db.ListObjects(ctx, metabase.ListObjects{
+						ProjectID:  baseObj.ProjectID,
+						BucketName: baseObj.BucketName,
+						Prefix:     "boo/",
+						Cursor:     metabase.ListObjectsCursor{Key: cursorKey},
+						Status:     metabase.Committed,
+						Limit:      batchsize,
+					})
+					require.NoError(b, err)
+
+				}
+			}
+		})
+	})
+}

satellite/metabase/metabasetest/test.go

@@ -322,6 +322,23 @@ func (step ListVerifySegments) Check(ctx *testcontext.Context, t testing.TB, db
 	require.Zero(t, diff)
 }
 
+// ListObjects is for testing metabase.ListObjects.
+type ListObjects struct {
+	Opts     metabase.ListObjects
+	Result   metabase.ListObjectsResult
+	ErrClass *errs.Class
+	ErrText  string
+}
+
+// Check runs the test.
+func (step ListObjects) Check(ctx *testcontext.Context, t testing.TB, db *metabase.DB) {
+	result, err := db.ListObjects(ctx, step.Opts)
+	checkError(t, err, step.ErrClass, step.ErrText)
+
+	diff := cmp.Diff(step.Result, result, DefaultTimeDiff(), cmpopts.EquateEmpty())
+	require.Zero(t, diff)
+}
+
 // ListStreamPositions is for testing metabase.ListStreamPositions.
 type ListStreamPositions struct {
 	Opts     metabase.ListStreamPositions