satellite/metabase: allow creating delete markers for missing objects
Change-Id: Ic8f6f80c234478b3e7b3bc4e88fbb5df9a783456
This commit is contained in:
parent
50c9f4c85a
commit
a44ed9457e
@ -420,37 +420,24 @@ func (db *DB) DeleteObjectLastCommitted(
|
||||
}
|
||||
|
||||
row := db.db.QueryRowContext(ctx, `
|
||||
WITH check_existing_object AS (
|
||||
SELECT status
|
||||
FROM objects
|
||||
WHERE
|
||||
(project_id, bucket_name, object_key) = ($1, $2, $3) AND
|
||||
status <> `+statusPending+`
|
||||
ORDER BY project_id, bucket_name, object_key, stream_id, version DESC, created_at DESC
|
||||
LIMIT 1
|
||||
),
|
||||
added_object AS (
|
||||
INSERT INTO objects (
|
||||
project_id, bucket_name, object_key, version, stream_id,
|
||||
status,
|
||||
zombie_deletion_deadline
|
||||
)
|
||||
SELECT
|
||||
$1, $2, $3,
|
||||
coalesce((
|
||||
SELECT version + 1
|
||||
FROM objects
|
||||
WHERE (project_id, bucket_name, object_key) = ($1, $2, $3)
|
||||
ORDER BY version DESC
|
||||
LIMIT 1
|
||||
), 1),
|
||||
$4,
|
||||
`+statusDeleteMarkerVersioned+`,
|
||||
NULL
|
||||
WHERE EXISTS (SELECT 1 FROM check_existing_object)
|
||||
RETURNING *
|
||||
INSERT INTO objects (
|
||||
project_id, bucket_name, object_key, version, stream_id,
|
||||
status,
|
||||
zombie_deletion_deadline
|
||||
)
|
||||
SELECT version, created_at FROM added_object
|
||||
SELECT
|
||||
$1, $2, $3,
|
||||
coalesce((
|
||||
SELECT version + 1
|
||||
FROM objects
|
||||
WHERE (project_id, bucket_name, object_key) = ($1, $2, $3)
|
||||
ORDER BY version DESC
|
||||
LIMIT 1
|
||||
), 1),
|
||||
$4,
|
||||
`+statusDeleteMarkerVersioned+`,
|
||||
NULL
|
||||
RETURNING version, created_at
|
||||
`, opts.ProjectID, []byte(opts.BucketName), opts.ObjectKey, deleterMarkerStreamID)
|
||||
|
||||
var deleted Object
|
||||
|
@ -452,18 +452,34 @@ func TestDeleteObjectVersioning(t *testing.T) {
|
||||
obj := metabasetest.RandObjectStream()
|
||||
location := obj.Location()
|
||||
|
||||
t.Run("Delete non existing object version", func(t *testing.T) {
|
||||
t.Run("delete non existing object version", func(t *testing.T) {
|
||||
defer metabasetest.DeleteAll{}.Check(ctx, t, db)
|
||||
|
||||
metabasetest.DeleteObjectLastCommitted{
|
||||
now := time.Now()
|
||||
marker := obj
|
||||
marker.Version = 1
|
||||
|
||||
result := metabasetest.DeleteObjectLastCommitted{
|
||||
Opts: metabase.DeleteObjectLastCommitted{
|
||||
ObjectLocation: location,
|
||||
Versioned: true,
|
||||
},
|
||||
ErrClass: &metabase.ErrObjectNotFound,
|
||||
Result: metabase.DeleteObjectResult{},
|
||||
Result: metabase.DeleteObjectResult{
|
||||
Markers: []metabase.Object{
|
||||
{
|
||||
ObjectStream: marker,
|
||||
CreatedAt: now,
|
||||
Status: metabase.DeleteMarkerVersioned,
|
||||
},
|
||||
},
|
||||
},
|
||||
}.Check(ctx, t, db)
|
||||
|
||||
metabasetest.Verify{
|
||||
Objects: []metabase.RawObject{
|
||||
metabase.RawObject(result.Markers[0]),
|
||||
},
|
||||
}.Check(ctx, t, db)
|
||||
metabasetest.Verify{}.Check(ctx, t, db)
|
||||
})
|
||||
|
||||
t.Run("Delete partial object", func(t *testing.T) {
|
||||
@ -476,26 +492,59 @@ func TestDeleteObjectVersioning(t *testing.T) {
|
||||
},
|
||||
}.Check(ctx, t, db)
|
||||
|
||||
metabasetest.DeleteObjectLastCommitted{
|
||||
now := time.Now()
|
||||
marker := obj
|
||||
marker.Version = pending.Version + 1
|
||||
|
||||
result := metabasetest.DeleteObjectLastCommitted{
|
||||
Opts: metabase.DeleteObjectLastCommitted{
|
||||
ObjectLocation: location,
|
||||
Versioned: true,
|
||||
},
|
||||
ErrClass: &metabase.ErrObjectNotFound,
|
||||
Result: metabase.DeleteObjectResult{},
|
||||
Result: metabase.DeleteObjectResult{
|
||||
Markers: []metabase.Object{
|
||||
{
|
||||
ObjectStream: marker,
|
||||
CreatedAt: now,
|
||||
Status: metabase.DeleteMarkerVersioned,
|
||||
},
|
||||
},
|
||||
},
|
||||
}.Check(ctx, t, db)
|
||||
|
||||
metabasetest.Verify{
|
||||
Objects: []metabase.RawObject{
|
||||
metabase.RawObject(pending),
|
||||
metabase.RawObject(result.Markers[0]),
|
||||
},
|
||||
}.Check(ctx, t, db)
|
||||
|
||||
marker2 := marker
|
||||
marker2.Version = marker.Version + 1
|
||||
result2 := metabasetest.DeleteObjectLastCommitted{
|
||||
Opts: metabase.DeleteObjectLastCommitted{
|
||||
ObjectLocation: location,
|
||||
Versioned: true,
|
||||
},
|
||||
Result: metabase.DeleteObjectResult{
|
||||
Markers: []metabase.Object{
|
||||
{
|
||||
ObjectStream: marker2,
|
||||
CreatedAt: now,
|
||||
Status: metabase.DeleteMarkerVersioned,
|
||||
},
|
||||
},
|
||||
},
|
||||
}.Check(ctx, t, db)
|
||||
|
||||
// Not quite sure whether this is the appropriate behavior,
|
||||
// but let's leave the pending object in place and not insert a deletion marker.
|
||||
metabasetest.Verify{
|
||||
Objects: []metabase.RawObject{{
|
||||
ObjectStream: obj,
|
||||
CreatedAt: pending.CreatedAt,
|
||||
Status: metabase.Pending,
|
||||
Encryption: pending.Encryption,
|
||||
|
||||
ZombieDeletionDeadline: pending.ZombieDeletionDeadline,
|
||||
}},
|
||||
Objects: []metabase.RawObject{
|
||||
metabase.RawObject(pending),
|
||||
metabase.RawObject(result.Markers[0]),
|
||||
metabase.RawObject(result2.Markers[0]),
|
||||
},
|
||||
}.Check(ctx, t, db)
|
||||
})
|
||||
|
||||
|
Loading…
Reference in New Issue
Block a user