Skip to content

Commit

Permalink
satellite/metabase: fix naive database behavior
Browse files Browse the repository at this point in the history 
This is relevant for scenarios, where the cursor positions itself
after the latest versions. In that scenario AllVersion=false
shouldn't list the lower versions.

Change-Id: I2d5fa853d6377b2708b1e8cb864da210b0a9b561
  • Loading branch information
@egonelbre
egonelbre committed Mar 14, 2024
1 parent 42d7317 commit adbf310
Show file tree
Hide file tree
Showing 3 changed files with 144 additions and 25 deletions.
5 changes: 5 additions & 0 deletions satellite/metabase/list_objects.go
View file
Expand Up @@ -358,3 +358,8 @@ func (db *DB) ListObjects(ctx context.Context, opts ListObjects) (result ListObj

// return entries, nil
// }

// VersionAscending returns whether the versions in the result are in ascending order.
func (opts *ListObjects) VersionAscending() bool {
return opts.Pending
}
12 changes: 12 additions & 0 deletions satellite/metabase/list_objects_exhaustive_test.go
View file
Expand Up @@ -53,6 +53,12 @@ func TestListObjects_Exhaustive(t *testing.T) {
opts.Cursor.Version = metabase.MaxVersion
check(opts)

opts.Cursor.Version = 4
check(opts)

opts.Cursor.Version = 5
check(opts)

for i := range entries {
entry := &entries[i]
opts.Cursor.Key = entry.ObjectKey
Expand All @@ -65,6 +71,12 @@ func TestListObjects_Exhaustive(t *testing.T) {

opts.Cursor.Version = metabase.MaxVersion
check(opts)

opts.Cursor.Version = entry.Version - 1
check(opts)

opts.Cursor.Version = entry.Version + 1
check(opts)
}
}
}
Expand Down
152 changes: 127 additions & 25 deletions satellite/metabase/list_objects_naive_test.go
View file
Expand Up @@ -18,21 +18,40 @@ import (

// NaiveObjectsDB implements a slow reference ListObjects implementation.
type NaiveObjectsDB struct {
VersionAsc []metabase.ObjectEntry
VersionDesc []metabase.ObjectEntry
VersionAsc []naiveObjectEntry
VersionDesc []naiveObjectEntry
}

type naiveObjectEntry struct {
IsLatest bool
metabase.ObjectEntry
}

// NewNaiveObjectsDB returns a new NaiveObjectsDB.
func NewNaiveObjectsDB(entries []metabase.ObjectEntry) *NaiveObjectsDB {
func NewNaiveObjectsDB(rawentries []metabase.ObjectEntry) *NaiveObjectsDB {
versiondesc := slices.Clone(rawentries)
sort.Slice(versiondesc, func(i, k int) bool {
return versiondesc[i].Less(versiondesc[k])
})

entries := make([]naiveObjectEntry, 0, len(rawentries))

lastEntry := metabase.ObjectKey("")
for _, entry := range versiondesc {
entries = append(entries, naiveObjectEntry{
IsLatest: entry.ObjectKey != lastEntry,
ObjectEntry: entry,
})
lastEntry = entry.ObjectKey
}

db := &NaiveObjectsDB{
VersionAsc: slices.Clone(entries),
VersionDesc: slices.Clone(entries),
VersionDesc: entries,
}

sort.Slice(db.VersionAsc, func(i, k int) bool {
return db.VersionAsc[i].LessVersionAsc(db.VersionAsc[k])
})
sort.Slice(db.VersionDesc, func(i, k int) bool {
return db.VersionDesc[i].Less(db.VersionDesc[k])
return db.VersionAsc[i].LessVersionAsc(db.VersionAsc[k].ObjectEntry)
})
return db
}
Expand Down Expand Up @@ -62,14 +81,19 @@ func (db *NaiveObjectsDB) ListObjects(ctx context.Context, opts metabase.ListObj
entryKeyWithPrefix, entryKey, entryVersion, isPrefix := calculateEntryKey(&opts, entry)

// The entry is before our cursor position.
if !lessIterateCursor(opts.Cursor, entryKeyWithPrefix, entryVersion) {
if entryExcludedByCursor(&opts, entryKeyWithPrefix, entryVersion, isPrefix) {
continue
}

if opts.Pending != (entry.Status == metabase.Pending) {
continue
}

// AllVersions=false should only care about the latest versions.
if !opts.AllVersions && !entry.IsLatest {
continue
}

if last != nil {
if isPrefix {
// prefix already included in output
Expand All @@ -90,7 +114,7 @@ func (db *NaiveObjectsDB) ListObjects(ctx context.Context, opts metabase.ListObj
scoped.IsPrefix = true
scoped.Status = metabase.Prefix
} else {
scoped = *entry
scoped = entry.ObjectEntry
scoped.ObjectKey = entryKey
clearEntryMetadata(&opts, &scoped)
}
Expand All @@ -115,14 +139,18 @@ func (db *NaiveObjectsDB) ListObjects(ctx context.Context, opts metabase.ListObj
}

// Less returns whether key and version are after the cursor.
func lessIterateCursor(cursor metabase.ListObjectsCursor, key metabase.ObjectKey, version metabase.Version) bool {
if cursor.Key == key {
return cursor.Version < version
func entryExcludedByCursor(opts *metabase.ListObjects, entryKeyWithPrefix metabase.ObjectKey, entryVersion metabase.Version, isPrefix bool) bool {
if opts.Cursor.Key == entryKeyWithPrefix && !isPrefix {
if opts.VersionAscending() {
return entryVersion <= opts.Cursor.Version
} else {
return entryVersion >= opts.Cursor.Version
}
}
return cursor.Key < key
return entryKeyWithPrefix <= opts.Cursor.Key
}

func calculateEntryKey(opts *metabase.ListObjects, entry *metabase.ObjectEntry) (entryKeyWithPrefix, entryKey metabase.ObjectKey, entryVersion metabase.Version, isPrefix bool) {
func calculateEntryKey(opts *metabase.ListObjects, entry *naiveObjectEntry) (entryKeyWithPrefix, entryKey metabase.ObjectKey, entryVersion metabase.Version, isPrefix bool) {
entryKey = entry.ObjectKey[len(opts.Prefix):]

if !opts.Recursive {
Expand Down Expand Up @@ -157,12 +185,11 @@ func clearEntryMetadata(opts *metabase.ListObjects, entry *metabase.ObjectEntry)

func TestNaiveObjectsDB_Basic(t *testing.T) {
check := func(entries []metabase.ObjectEntry, opts metabase.ListObjects, expected []metabase.ObjectEntry) {
t.Run("", func(t *testing.T) {
naive := NewNaiveObjectsDB(entries)
result, err := naive.ListObjects(context.Background(), opts)
require.NoError(t, err)
require.Equal(t, expected, result.Objects)
})
t.Helper()
naive := NewNaiveObjectsDB(entries)
result, err := naive.ListObjects(context.Background(), opts)
require.NoError(t, err)
require.Equal(t, expected, result.Objects)
}

check(
Expand All @@ -172,13 +199,88 @@ func TestNaiveObjectsDB_Basic(t *testing.T) {
{ObjectKey: "b", Version: 1, Status: metabase.CommittedVersioned},
},
metabase.ListObjects{
Recursive: false,
Limit: 2,
Prefix: "",
Cursor: metabase.ListObjectsCursor{Key: "a/", Version: 0},
AllVersions: false,
Recursive: false,
Pending: false,
Limit: 2,
Prefix: "",
Cursor: metabase.ListObjectsCursor{Key: "a/", Version: 0},
},
[]metabase.ObjectEntry{
{ObjectKey: "b", Version: 1, Status: metabase.CommittedVersioned},
},
)

check(
[]metabase.ObjectEntry{
{ObjectKey: "\x00", Version: 3, Status: metabase.CommittedVersioned},
{ObjectKey: "\x00\x00", Version: 1, Status: metabase.CommittedVersioned},
},
metabase.ListObjects{
AllVersions: true,
Recursive: true,
Pending: false,
Limit: 1,
Prefix: "",
Cursor: metabase.ListObjectsCursor{Key: "\x00", Version: 0},
},
[]metabase.ObjectEntry{
{ObjectKey: "\x00\x00", Version: 1, Status: metabase.CommittedVersioned},
},
)

check(
[]metabase.ObjectEntry{
{ObjectKey: "\x00/", Version: 10, Status: metabase.CommittedVersioned},
{ObjectKey: "\x00\xff", Version: 5, Status: metabase.CommittedVersioned},
},
metabase.ListObjects{
AllVersions: true,
Recursive: false,
Pending: false,
Limit: 1,
Prefix: "",
Cursor: metabase.ListObjectsCursor{Key: "\x00/", Version: 0},
},
[]metabase.ObjectEntry{
{ObjectKey: "\x00\xff", Version: 5, Status: metabase.CommittedVersioned},
},
)

check(
[]metabase.ObjectEntry{
{ObjectKey: "\x00/", Version: 10, Status: metabase.CommittedVersioned},
{ObjectKey: "\x00\xff", Version: 5, Status: metabase.CommittedVersioned},
},
metabase.ListObjects{
AllVersions: true,
Recursive: false,
Pending: false,
Limit: 2,
Prefix: "",
Cursor: metabase.ListObjectsCursor{Key: "\x00/", Version: metabase.MaxVersion},
},
[]metabase.ObjectEntry{
{ObjectKey: "\x00\xff", Version: 5, Status: metabase.CommittedVersioned},
},
)

check(
[]metabase.ObjectEntry{
{ObjectKey: "\x00/", Version: 10, Status: metabase.CommittedVersioned},
{ObjectKey: "\x00\xff", Version: 5, Status: metabase.CommittedVersioned},
},
metabase.ListObjects{
AllVersions: true,
Recursive: false,
Pending: false,
Limit: 2,
Prefix: "",
Cursor: metabase.ListObjectsCursor{Key: "\x00", Version: 0},
},
[]metabase.ObjectEntry{
{ObjectKey: "\x00/", Version: 0, Status: metabase.Prefix, IsPrefix: true},
{ObjectKey: "\x00\xff", Version: 5, Status: metabase.CommittedVersioned},
},
)
}

0 comments on commit adbf310

Please sign in to comment.