diff --git a/satellite/metabase/iterator_test.go b/satellite/metabase/iterator_test.go index 13e600bf9..78c815631 100644 --- a/satellite/metabase/iterator_test.go +++ b/satellite/metabase/iterator_test.go @@ -972,6 +972,956 @@ func TestIterateObjectsWithStatus(t *testing.T) { }, }.Check(ctx, t, db) }) + + t.Run("2 objects, one with multiple versions and one without versioning", func(t *testing.T) { + defer metabasetest.DeleteAll{}.Check(ctx, t, db) + + a0 := metabasetest.RandObjectStream() + b0 := metabasetest.RandObjectStream() + b0.ProjectID = a0.ProjectID + b0.BucketName = a0.BucketName + b0.Version = 1000 + + if a0.ObjectKey > b0.ObjectKey { + b0.ObjectKey, a0.ObjectKey = a0.ObjectKey, b0.ObjectKey + } + + b1 := b0 + b1.Version = 500 + + objA0 := metabasetest.CreateObject(ctx, t, db, a0, 0) + objB0 := metabasetest.CreateObjectVersioned(ctx, t, db, b0, 0) + objB1 := metabasetest.CreateObjectVersionedOutOfOrder(ctx, t, db, b1, 0, 1001) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: a0.ProjectID, + BucketName: a0.BucketName, + Recursive: true, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + }, + Result: []metabase.ObjectEntry{ + objectEntryFromRaw(metabase.RawObject(objA0)), + objectEntryFromRaw(metabase.RawObject(objB0)), + objectEntryFromRaw(metabase.RawObject(objB1)), + }, + }.Check(ctx, t, db) + + metabasetest.Verify{ + Objects: []metabase.RawObject{ + metabase.RawObject(objA0), + metabase.RawObject(objB0), + metabase.RawObject(objB1), + }, + }.Check(ctx, t, db) + }) + + t.Run("3 objects, one with versions one without and one pending", func(t *testing.T) { + defer metabasetest.DeleteAll{}.Check(ctx, t, db) + + a0 := metabasetest.RandObjectStream() + b0 := metabasetest.RandObjectStream() + c0 := metabasetest.RandObjectStream() + b0.ProjectID = a0.ProjectID + b0.BucketName = a0.BucketName + b0.Version = 1000 + c0.ProjectID = a0.ProjectID + c0.BucketName = a0.BucketName + c0.Version = 1000 + + if a0.ObjectKey > b0.ObjectKey { + b0.ObjectKey, a0.ObjectKey = a0.ObjectKey, b0.ObjectKey + } + + b1 := b0 + b1.Version = 500 + + objA0 := metabasetest.CreateObject(ctx, t, db, a0, 0) + objB0 := metabasetest.CreateObjectVersioned(ctx, t, db, b0, 0) + objB1 := metabasetest.CreateObjectVersionedOutOfOrder(ctx, t, db, b1, 0, 1001) + metabasetest.CreatePendingObject(ctx, t, db, c0, 0) + now := time.Now() + zombieDeadline := now.Add(24 * time.Hour) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: a0.ProjectID, + BucketName: a0.BucketName, + Recursive: true, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + }, + Result: []metabase.ObjectEntry{ + objectEntryFromRaw(metabase.RawObject(objA0)), + objectEntryFromRaw(metabase.RawObject(objB0)), + objectEntryFromRaw(metabase.RawObject(objB1)), + }, + }.Check(ctx, t, db) + + metabasetest.Verify{ + Objects: []metabase.RawObject{ + metabase.RawObject(objA0), + metabase.RawObject(objB0), + metabase.RawObject(objB1), + { + ObjectStream: metabase.ObjectStream{ + ProjectID: c0.ProjectID, + BucketName: c0.BucketName, + ObjectKey: c0.ObjectKey, + Version: 1000, + StreamID: c0.StreamID, + }, + CreatedAt: now, + Status: metabase.Pending, + + Encryption: metabasetest.DefaultEncryption, + ZombieDeletionDeadline: &zombieDeadline, + }, + }, + }.Check(ctx, t, db) + }) + + t.Run("2 objects one with versions and one pending, list pending", func(t *testing.T) { + defer metabasetest.DeleteAll{}.Check(ctx, t, db) + + a0 := metabasetest.RandObjectStream() + a0.Version = 1000 + b0 := metabasetest.RandObjectStream() + b0.ProjectID = a0.ProjectID + b0.BucketName = a0.BucketName + b0.Version = 1000 + + if a0.ObjectKey > b0.ObjectKey { + b0.ObjectKey, a0.ObjectKey = a0.ObjectKey, b0.ObjectKey + } + + a1 := a0 + a1.Version = 1001 + + metabasetest.BeginObjectExactVersion{ + Opts: metabase.BeginObjectExactVersion{ + ObjectStream: metabase.ObjectStream{ + ProjectID: b0.ProjectID, + BucketName: b0.BucketName, + ObjectKey: b0.ObjectKey, + Version: b0.Version, + StreamID: b0.StreamID, + }, + Encryption: metabasetest.DefaultEncryption, + }, + }.Check(ctx, t, db) + now := time.Now() + zombieDeadline := now.Add(24 * time.Hour) + + objA0 := metabasetest.CreateObjectVersioned(ctx, t, db, a0, 0) + objA1 := metabasetest.CreateObjectVersioned(ctx, t, db, a1, 0) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: a0.ProjectID, + BucketName: a0.BucketName, + Recursive: true, + Pending: true, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + }, + Result: []metabase.ObjectEntry{ + { + ObjectKey: b0.ObjectKey, + Version: 1000, + StreamID: b0.StreamID, + CreatedAt: now, + Status: metabase.Pending, + + Encryption: metabasetest.DefaultEncryption, + }, + }, + }.Check(ctx, t, db) + + metabasetest.Verify{ + Objects: []metabase.RawObject{ + metabase.RawObject(objA0), + metabase.RawObject(objA1), + { + ObjectStream: metabase.ObjectStream{ + ProjectID: b0.ProjectID, + BucketName: b0.BucketName, + ObjectKey: b0.ObjectKey, + Version: 1000, + StreamID: b0.StreamID, + }, + CreatedAt: now, + Status: metabase.Pending, + + Encryption: metabasetest.DefaultEncryption, + ZombieDeletionDeadline: &zombieDeadline, + }, + }, + }.Check(ctx, t, db) + }) + + t.Run("2 objects, each with 2 versions", func(t *testing.T) { + defer metabasetest.DeleteAll{}.Check(ctx, t, db) + + a0 := metabasetest.RandObjectStream() + a0.Version = 1000 + b0 := metabasetest.RandObjectStream() + b0.ProjectID = a0.ProjectID + b0.BucketName = a0.BucketName + b0.Version = 1000 + + if a0.ObjectKey > b0.ObjectKey { + b0.ObjectKey, a0.ObjectKey = a0.ObjectKey, b0.ObjectKey + } + + a1 := a0 + a1.Version = 1001 + b1 := b0 + b1.Version = 500 + + objA0 := metabasetest.CreateObjectVersioned(ctx, t, db, a0, 0) + objA1 := metabasetest.CreateObjectVersioned(ctx, t, db, a1, 0) + objB0 := metabasetest.CreateObjectVersioned(ctx, t, db, b0, 0) + objB1 := metabasetest.CreateObjectVersionedOutOfOrder(ctx, t, db, b1, 0, 1001) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: a0.ProjectID, + BucketName: a0.BucketName, + Recursive: true, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + }, + Result: []metabase.ObjectEntry{ + objectEntryFromRaw(metabase.RawObject(objA0)), + objectEntryFromRaw(metabase.RawObject(objA1)), + objectEntryFromRaw(metabase.RawObject(objB0)), + objectEntryFromRaw(metabase.RawObject(objB1)), + }}.Check(ctx, t, db) + + metabasetest.Verify{ + Objects: []metabase.RawObject{ + metabase.RawObject(objA0), + metabase.RawObject(objA1), + metabase.RawObject(objB0), + metabase.RawObject(objB1), + }, + }.Check(ctx, t, db) + }) + + t.Run("2 objects, each with two versions and one with delete_marker", func(t *testing.T) { + defer metabasetest.DeleteAll{}.Check(ctx, t, db) + + a0 := metabasetest.RandObjectStream() + a0.Version = 1000 + b0 := metabasetest.RandObjectStream() + b0.ProjectID = a0.ProjectID + b0.BucketName = a0.BucketName + b0.Version = 1000 + + if a0.ObjectKey > b0.ObjectKey { + b0.ObjectKey, a0.ObjectKey = a0.ObjectKey, b0.ObjectKey + } + + a1 := a0 + a1.Version = 1001 + b1 := b0 + b1.Version = 500 + + objA0 := metabasetest.CreateObjectVersioned(ctx, t, db, a0, 0) + objA1 := metabasetest.CreateObjectVersioned(ctx, t, db, a1, 0) + objB0 := metabasetest.CreateObjectVersioned(ctx, t, db, b0, 0) + objB1 := metabasetest.CreateObjectVersionedOutOfOrder(ctx, t, db, b1, 0, 1001) + + deletionResult := metabasetest.DeleteObjectLastCommitted{ + Opts: metabase.DeleteObjectLastCommitted{ + ObjectLocation: objA0.Location(), + Versioned: true, + }, + Result: metabase.DeleteObjectResult{ + Markers: []metabase.Object{ + { + ObjectStream: metabase.ObjectStream{ + ProjectID: objA0.ProjectID, + BucketName: objA0.BucketName, + ObjectKey: objA0.ObjectKey, + Version: 1002, + }, + Status: metabase.DeleteMarkerVersioned, + CreatedAt: time.Now(), + }, + }, + }, + }.Check(ctx, t, db) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: a0.ProjectID, + BucketName: a0.BucketName, + Recursive: true, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + }, + Result: []metabase.ObjectEntry{ + objectEntryFromRaw(metabase.RawObject(objA0)), + objectEntryFromRaw(metabase.RawObject(objA1)), + objectEntryFromRaw(metabase.RawObject(deletionResult.Markers[0])), + objectEntryFromRaw(metabase.RawObject(objB0)), + objectEntryFromRaw(metabase.RawObject(objB1)), + }}.Check(ctx, t, db) + + metabasetest.Verify{ + Objects: []metabase.RawObject{ + metabase.RawObject(deletionResult.Markers[0]), + metabase.RawObject(objA0), + metabase.RawObject(objA1), + metabase.RawObject(objB0), + metabase.RawObject(objB1), + }, + }.Check(ctx, t, db) + }) + + t.Run("2 objects, each with two versions and multiple delete_markers", func(t *testing.T) { + defer metabasetest.DeleteAll{}.Check(ctx, t, db) + + a0 := metabasetest.RandObjectStream() + a0.Version = 1000 + b0 := metabasetest.RandObjectStream() + b0.ProjectID = a0.ProjectID + b0.BucketName = a0.BucketName + b0.Version = 1000 + + if a0.ObjectKey > b0.ObjectKey { + b0.ObjectKey, a0.ObjectKey = a0.ObjectKey, b0.ObjectKey + } + + objA0 := metabasetest.CreateObjectVersioned(ctx, t, db, a0, 0) + objB0 := metabasetest.CreateObjectVersioned(ctx, t, db, b0, 0) + + deletionResultA0 := metabasetest.DeleteObjectLastCommitted{ + Opts: metabase.DeleteObjectLastCommitted{ + ObjectLocation: objA0.Location(), + Versioned: true, + }, + Result: metabase.DeleteObjectResult{ + Markers: []metabase.Object{ + { + ObjectStream: metabase.ObjectStream{ + ProjectID: objA0.ProjectID, + BucketName: objA0.BucketName, + ObjectKey: objA0.ObjectKey, + Version: 1001, + }, + Status: metabase.DeleteMarkerVersioned, + CreatedAt: time.Now(), + }, + }, + }, + }.Check(ctx, t, db) + + deletionResultB0 := metabasetest.DeleteObjectLastCommitted{ + Opts: metabase.DeleteObjectLastCommitted{ + ObjectLocation: objB0.Location(), + Versioned: true, + }, + Result: metabase.DeleteObjectResult{ + Markers: []metabase.Object{ + { + ObjectStream: metabase.ObjectStream{ + ProjectID: objB0.ProjectID, + BucketName: objB0.BucketName, + ObjectKey: objB0.ObjectKey, + Version: 1001, + }, + Status: metabase.DeleteMarkerVersioned, + CreatedAt: time.Now(), + }, + }, + }, + }.Check(ctx, t, db) + + a1 := a0 + a1.Version = 1002 + b1 := b0 + b1.Version = 1002 + + objA1 := metabasetest.CreateObjectVersioned(ctx, t, db, a1, 0) + objB1 := metabasetest.CreateObjectVersioned(ctx, t, db, b1, 0) + + deletionResultA1 := metabasetest.DeleteObjectLastCommitted{ + Opts: metabase.DeleteObjectLastCommitted{ + ObjectLocation: objA1.Location(), + Versioned: true, + }, + Result: metabase.DeleteObjectResult{ + Markers: []metabase.Object{ + { + ObjectStream: metabase.ObjectStream{ + ProjectID: objA1.ProjectID, + BucketName: objA1.BucketName, + ObjectKey: objA1.ObjectKey, + Version: 1003, + }, + Status: metabase.DeleteMarkerVersioned, + CreatedAt: time.Now(), + }, + }, + }, + }.Check(ctx, t, db) + + deletionResultB1 := metabasetest.DeleteObjectLastCommitted{ + Opts: metabase.DeleteObjectLastCommitted{ + ObjectLocation: objB1.Location(), + Versioned: true, + }, + Result: metabase.DeleteObjectResult{ + Markers: []metabase.Object{ + { + ObjectStream: metabase.ObjectStream{ + ProjectID: objB1.ProjectID, + BucketName: objB1.BucketName, + ObjectKey: objB1.ObjectKey, + Version: 1003, + }, + Status: metabase.DeleteMarkerVersioned, + CreatedAt: time.Now(), + }, + }, + }, + }.Check(ctx, t, db) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: a0.ProjectID, + BucketName: a0.BucketName, + Recursive: true, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + }, + Result: []metabase.ObjectEntry{ + objectEntryFromRaw(metabase.RawObject(objA0)), + objectEntryFromRaw(metabase.RawObject(deletionResultA0.Markers[0])), + objectEntryFromRaw(metabase.RawObject(objA1)), + objectEntryFromRaw(metabase.RawObject(deletionResultA1.Markers[0])), + objectEntryFromRaw(metabase.RawObject(objB0)), + objectEntryFromRaw(metabase.RawObject(deletionResultB0.Markers[0])), + objectEntryFromRaw(metabase.RawObject(objB1)), + objectEntryFromRaw(metabase.RawObject(deletionResultB1.Markers[0])), + }}.Check(ctx, t, db) + + metabasetest.Verify{ + Objects: []metabase.RawObject{ + metabase.RawObject(deletionResultA1.Markers[0]), + metabase.RawObject(deletionResultB1.Markers[0]), + metabase.RawObject(objA1), + metabase.RawObject(objB1), + metabase.RawObject(deletionResultA0.Markers[0]), + metabase.RawObject(deletionResultB0.Markers[0]), + metabase.RawObject(objA0), + metabase.RawObject(objB0), + }, + }.Check(ctx, t, db) + }) + + t.Run("3 objects, 1 unversioned, 2 with multiple versions, 1 with and 1 without delete_marker", func(t *testing.T) { + defer metabasetest.DeleteAll{}.Check(ctx, t, db) + + a0 := metabasetest.RandObjectStream() + b0 := metabasetest.RandObjectStream() + b0.ProjectID = a0.ProjectID + b0.BucketName = a0.BucketName + b0.Version = 1000 + c0 := metabasetest.RandObjectStream() + c0.ProjectID = a0.ProjectID + c0.BucketName = a0.BucketName + c0.Version = 1000 + + if a0.ObjectKey > b0.ObjectKey { + a0.ObjectKey, b0.ObjectKey = b0.ObjectKey, a0.ObjectKey + } + if a0.ObjectKey > c0.ObjectKey { + a0.ObjectKey, c0.ObjectKey = c0.ObjectKey, a0.ObjectKey + } + if b0.ObjectKey > c0.ObjectKey { + b0.ObjectKey, c0.ObjectKey = c0.ObjectKey, b0.ObjectKey + } + + objA0 := metabasetest.CreateObject(ctx, t, db, a0, 0) + objB0 := metabasetest.CreateObjectVersioned(ctx, t, db, b0, 0) + objC0 := metabasetest.CreateObjectVersioned(ctx, t, db, c0, 0) + + deletionResultC0 := metabasetest.DeleteObjectLastCommitted{ + Opts: metabase.DeleteObjectLastCommitted{ + ObjectLocation: objC0.Location(), + Versioned: true, + }, + Result: metabase.DeleteObjectResult{ + Markers: []metabase.Object{ + { + ObjectStream: metabase.ObjectStream{ + ProjectID: objC0.ProjectID, + BucketName: objC0.BucketName, + ObjectKey: objC0.ObjectKey, + Version: 1001, + }, + Status: metabase.DeleteMarkerVersioned, + CreatedAt: time.Now(), + }, + }, + }, + }.Check(ctx, t, db) + + b1 := b0 + b1.Version = 1001 + c1 := c0 + c1.Version = 1002 + + objB1 := metabasetest.CreateObjectVersioned(ctx, t, db, b1, 0) + objC1 := metabasetest.CreateObjectVersioned(ctx, t, db, c1, 0) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: a0.ProjectID, + BucketName: a0.BucketName, + Recursive: true, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + }, + Result: []metabase.ObjectEntry{ + objectEntryFromRaw(metabase.RawObject(objA0)), + objectEntryFromRaw(metabase.RawObject(objB0)), + objectEntryFromRaw(metabase.RawObject(objB1)), + objectEntryFromRaw(metabase.RawObject(objC0)), + objectEntryFromRaw(metabase.RawObject(deletionResultC0.Markers[0])), + objectEntryFromRaw(metabase.RawObject(objC1)), + }, + }.Check(ctx, t, db) + + metabasetest.Verify{ + Objects: []metabase.RawObject{ + metabase.RawObject(objC1), + metabase.RawObject(objB1), + metabase.RawObject(deletionResultC0.Markers[0]), + metabase.RawObject(objC0), + metabase.RawObject(objB0), + metabase.RawObject(objA0), + }, + }.Check(ctx, t, db) + }) + + t.Run("list recursive objects with versions", func(t *testing.T) { + defer metabasetest.DeleteAll{}.Check(ctx, t, db) + projectID, bucketName := uuid.UUID{1}, "bucky" + + objects := metabasetest.CreateVersionedObjectsWithKeysAll(ctx, t, db, projectID, bucketName, map[metabase.ObjectKey][]metabase.Version{ + "a": {1000, 1001}, + "b/1": {1000, 1001}, + "b/2": {1000, 1001}, + "b/3": {1000, 1001}, + "c": {1000, 1001}, + "c/": {1000, 1001}, + "c//": {1000, 1001}, + "c/1": {1000, 1001}, + "g": {1000, 1001}, + }) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: projectID, + BucketName: bucketName, + Recursive: true, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + }, + Result: concat( + 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.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: projectID, + BucketName: bucketName, + Recursive: true, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + + Cursor: metabase.IterateCursor{Key: "a", Version: last(objects["a"]).Version + 1}, + }, + Result: concat( + 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.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: projectID, + BucketName: bucketName, + Recursive: true, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + + Cursor: metabase.IterateCursor{Key: "b", Version: 0}, + }, + Result: concat( + 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.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: projectID, + BucketName: bucketName, + Recursive: true, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + + Prefix: "b/", + }, + Result: withoutPrefix("b/", + concat( + objects["b/1"], + objects["b/2"], + objects["b/3"], + )..., + ), + }.Check(ctx, t, db) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: projectID, + BucketName: bucketName, + Recursive: true, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + + Prefix: "b/", + Cursor: metabase.IterateCursor{Key: "a"}, + }, + Result: withoutPrefix("b/", + concat( + objects["b/1"], + objects["b/2"], + objects["b/3"], + )..., + ), + }.Check(ctx, t, db) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: projectID, + BucketName: bucketName, + Recursive: true, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + + Prefix: "b/", + Cursor: metabase.IterateCursor{Key: "b/2", Version: -3}, + }, + Result: withoutPrefix("b/", + concat( + objects["b/2"], + objects["b/3"], + )..., + ), + }.Check(ctx, t, db) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: projectID, + BucketName: bucketName, + Recursive: true, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + + Prefix: "b/", + Cursor: metabase.IterateCursor{Key: "c/"}, + }, + Result: nil, + }.Check(ctx, t, db) + }) + + t.Run("list non-recursive objects with versions", func(t *testing.T) { + defer metabasetest.DeleteAll{}.Check(ctx, t, db) + projectID, bucketName := uuid.UUID{1}, "bucky" + + objects := metabasetest.CreateVersionedObjectsWithKeysAll(ctx, t, db, projectID, bucketName, map[metabase.ObjectKey][]metabase.Version{ + "a": {1000, 1001}, + "b/1": {1000, 1001}, + "b/2": {1000, 1001}, + "b/3": {1000, 1001}, + "c": {1000, 1001}, + "c/": {1000, 1001}, + "c//": {1000, 1001}, + "c/1": {1000, 1001}, + "g": {1000, 1001}, + }) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: projectID, + BucketName: bucketName, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + }, + Result: concat( + objects["a"], + []metabase.ObjectEntry{prefixEntry("b/")}, + objects["c"], + []metabase.ObjectEntry{prefixEntry("c/")}, + objects["g"], + ), + }.Check(ctx, t, db) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: projectID, + BucketName: bucketName, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + + Cursor: metabase.IterateCursor{Key: "a", Version: last(objects["a"]).Version + 1}, + }, + Result: concat( + []metabase.ObjectEntry{prefixEntry("b/")}, + objects["c"], + []metabase.ObjectEntry{prefixEntry("c/")}, + objects["g"], + ), + }.Check(ctx, t, db) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: projectID, + BucketName: bucketName, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + + Cursor: metabase.IterateCursor{Key: "b", Version: 0}, + }, + Result: concat( + []metabase.ObjectEntry{prefixEntry("b/")}, + objects["c"], + []metabase.ObjectEntry{prefixEntry("c/")}, + objects["g"], + ), + }.Check(ctx, t, db) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: projectID, + BucketName: bucketName, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + + Prefix: "b/", + }, + Result: withoutPrefix("b/", + concat( + objects["b/1"], + objects["b/2"], + objects["b/3"], + )..., + ), + }.Check(ctx, t, db) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: projectID, + BucketName: bucketName, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + + Prefix: "b/", + Cursor: metabase.IterateCursor{Key: "a"}, + }, + Result: withoutPrefix("b/", + concat( + objects["b/1"], + objects["b/2"], + objects["b/3"], + )..., + ), + }.Check(ctx, t, db) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: projectID, + BucketName: bucketName, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + + Prefix: "b/", + Cursor: metabase.IterateCursor{Key: "b/2", Version: -3}, + }, + Result: withoutPrefix("b/", + concat( + objects["b/2"], + objects["b/3"], + )..., + ), + }.Check(ctx, t, db) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: projectID, + BucketName: bucketName, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + + Prefix: "b/", + Cursor: metabase.IterateCursor{Key: "c/"}, + }, + Result: nil, + }.Check(ctx, t, db) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: projectID, + BucketName: bucketName, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + + Prefix: "c/", + Cursor: metabase.IterateCursor{Key: "c/"}, + }, + Result: withoutPrefix("c/", + concat( + objects["c/"], + []metabase.ObjectEntry{prefixEntry("c//")}, + objects["c/1"], + )..., + ), + }.Check(ctx, t, db) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: projectID, + BucketName: bucketName, + Pending: false, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + + Prefix: "c//", + }, + Result: withoutPrefix("c//", + objects["c//"]..., + ), + }.Check(ctx, t, db) + }) + + t.Run("batch iterate committed versioned, unversioned, and delete markers with pending object", func(t *testing.T) { + defer metabasetest.DeleteAll{}.Check(ctx, t, db) + now := time.Now() + zombieDeadline := now.Add(24 * time.Hour) + + var expected []metabase.ObjectEntry + var objLocation metabase.ObjectLocation + + // create 1 pending object first + pendingStream1 := metabasetest.RandObjectStream() + objLocation = pendingStream1.Location() + pendingStream1.Version = 100 + + metabasetest.CreatePendingObject(ctx, t, db, pendingStream1, 0) + + pendingObject1 := metabase.RawObject{ + ObjectStream: pendingStream1, + CreatedAt: now, + Status: metabase.Pending, + Encryption: metabasetest.DefaultEncryption, + ZombieDeletionDeadline: &zombieDeadline, + } + expected = append(expected, objectEntryFromRaw(pendingObject1)) + + for i := 0; i < 10; i++ { + unversionedStream := metabasetest.RandObjectStream() + unversionedStream.ProjectID = objLocation.ProjectID + unversionedStream.BucketName = objLocation.BucketName + unversionedStream.ObjectKey = objLocation.ObjectKey + unversionedStream.Version = metabase.Version(200 + i) + if i == 0 { + metabasetest.CreateObject(ctx, t, db, unversionedStream, 0) + } else { + metabasetest.CreateObjectVersioned(ctx, t, db, unversionedStream, 0) + } + } + + // create a second pending object + pendingStream2 := metabasetest.RandObjectStream() + pendingStream2.ProjectID = objLocation.ProjectID + pendingStream2.BucketName = objLocation.BucketName + pendingStream2.ObjectKey = objLocation.ObjectKey + pendingStream2.Version = 300 + + metabasetest.CreatePendingObject(ctx, t, db, pendingStream2, 0) + + pendingObject2 := metabase.RawObject{ + ObjectStream: pendingStream2, + CreatedAt: now, + Status: metabase.Pending, + Encryption: metabasetest.DefaultEncryption, + ZombieDeletionDeadline: &zombieDeadline, + } + expected = append(expected, objectEntryFromRaw(pendingObject2)) + + metabasetest.IterateObjectsWithStatus{ + Opts: metabase.IterateObjectsWithStatus{ + ProjectID: objLocation.ProjectID, + BucketName: objLocation.BucketName, + Pending: true, + IncludeCustomMetadata: true, + IncludeSystemMetadata: true, + BatchSize: 3, + Recursive: true, + }, + Result: expected, + }.Check(ctx, t, db) + }) }) } @@ -1763,3 +2713,14 @@ func BenchmarkNonRecursiveListing(b *testing.B) { }) }) } + +func concat[E any](slices ...[]E) (concatenated []E) { + for _, s := range slices { + concatenated = append(concatenated, s...) + } + return concatenated +} + +func last[E any](someSlice []E) E { + return someSlice[len(someSlice)-1] +} diff --git a/satellite/metabase/metabasetest/create.go b/satellite/metabase/metabasetest/create.go index 80a149faa..f34f71c13 100644 --- a/satellite/metabase/metabasetest/create.go +++ b/satellite/metabase/metabasetest/create.go @@ -216,7 +216,8 @@ func CreateSegments(ctx *testcontext.Context, t testing.TB, db *metabase.DB, obj } } -// CreateVersionedObjectsWithKeys creates multiple versioned objects with the specified keys and versions. +// CreateVersionedObjectsWithKeys creates multiple versioned objects with the specified keys and versions, +// and returns a mapping of keys to final versions. func CreateVersionedObjectsWithKeys(ctx *testcontext.Context, t *testing.T, db *metabase.DB, projectID uuid.UUID, bucketName string, keys map[metabase.ObjectKey][]metabase.Version) map[metabase.ObjectKey]metabase.ObjectEntry { objects := make(map[metabase.ObjectKey]metabase.ObjectEntry, len(keys)) for key, versions := range keys { @@ -244,6 +245,35 @@ func CreateVersionedObjectsWithKeys(ctx *testcontext.Context, t *testing.T, db * return objects } +// CreateVersionedObjectsWithKeysAll creates multiple versioned objects with the specified keys and versions, +// and returns a mapping of keys to a slice of all versions. +func CreateVersionedObjectsWithKeysAll(ctx *testcontext.Context, t *testing.T, db *metabase.DB, projectID uuid.UUID, bucketName string, keys map[metabase.ObjectKey][]metabase.Version) map[metabase.ObjectKey][]metabase.ObjectEntry { + objects := make(map[metabase.ObjectKey][]metabase.ObjectEntry, len(keys)) + for key, versions := range keys { + for _, version := range versions { + obj := RandObjectStream() + obj.ProjectID = projectID + obj.BucketName = bucketName + obj.ObjectKey = key + obj.Version = version + now := time.Now() + + CreateObjectVersioned(ctx, t, db, obj, 0) + + objects[key] = append(objects[key], metabase.ObjectEntry{ + ObjectKey: obj.ObjectKey, + Version: obj.Version, + StreamID: obj.StreamID, + CreatedAt: now, + Status: metabase.CommittedVersioned, + Encryption: DefaultEncryption, + }) + } + } + + return objects +} + // CreateTestObject is for testing metabase.CreateTestObject. type CreateTestObject struct { BeginObjectExactVersion *metabase.BeginObjectExactVersion