satellite/metainfo/metabase: add GetLatestObjectLastSegment request

Change-Id: I50c452e379f78864b38921f31e53a0554642ab02
This commit is contained in:
Michal Niewrzal 2020-11-03 16:03:20 +01:00
parent f986fdfe72
commit 2398afe985
3 changed files with 148 additions and 0 deletions

View File

@ -196,3 +196,53 @@ func (db *DB) GetSegmentByPosition(ctx context.Context, opts GetSegmentByPositio
return segment, nil return segment, nil
} }
// GetLatestObjectLastSegment contains arguments necessary for fetching a last segment information.
type GetLatestObjectLastSegment struct {
ObjectLocation
}
// GetLatestObjectLastSegment returns an object last segment information.
func (db *DB) GetLatestObjectLastSegment(ctx context.Context, opts GetLatestObjectLastSegment) (segment Segment, err error) {
defer mon.Task()(&ctx)(&err)
if err := opts.Verify(); err != nil {
return Segment{}, err
}
err = db.db.QueryRow(ctx, `
SELECT
stream_id, position,
root_piece_id, encrypted_key_nonce, encrypted_key,
encrypted_size, plain_offset, plain_size,
redundancy,
inline_data, remote_pieces
FROM segments
WHERE
stream_id = (SELECT stream_id FROM objects WHERE
project_id = $1 AND
bucket_name = $2 AND
object_key = $3 AND
status = 1
ORDER BY version DESC
LIMIT 1
)
ORDER BY position DESC
LIMIT 1
`, opts.ProjectID, opts.BucketName, []byte(opts.ObjectKey)).
Scan(
&segment.StreamID, &segment.Position,
&segment.RootPieceID, &segment.EncryptedKeyNonce, &segment.EncryptedKey,
&segment.EncryptedSize, &segment.PlainOffset, &segment.PlainSize,
redundancyScheme{&segment.Redundancy},
&segment.InlineData, &segment.Pieces,
)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return Segment{}, storj.ErrObjectNotFound.Wrap(Error.New("object or segment missing"))
}
return Segment{}, Error.New("unable to query segment: %w", err)
}
return segment, nil
}

View File

@ -407,3 +407,86 @@ func TestGetSegmentByPosition(t *testing.T) {
}) })
}) })
} }
func TestGetLatestObjectLastSegment(t *testing.T) {
All(t, func(ctx *testcontext.Context, t *testing.T, db *metabase.DB) {
obj := randObjectStream()
location := obj.Location()
now := time.Now()
for _, test := range invalidObjectLocations(location) {
test := test
t.Run(test.Name, func(t *testing.T) {
defer DeleteAll{}.Check(ctx, t, db)
GetLatestObjectLastSegment{
Opts: metabase.GetLatestObjectLastSegment{
ObjectLocation: test.ObjectLocation,
},
ErrClass: test.ErrClass,
ErrText: test.ErrText,
}.Check(ctx, t, db)
Verify{}.Check(ctx, t, db)
})
}
t.Run("Object or segment missing", func(t *testing.T) {
defer DeleteAll{}.Check(ctx, t, db)
GetLatestObjectLastSegment{
Opts: metabase.GetLatestObjectLastSegment{
ObjectLocation: location,
},
ErrClass: &storj.ErrObjectNotFound,
ErrText: "metabase: object or segment missing",
}.Check(ctx, t, db)
Verify{}.Check(ctx, t, db)
})
t.Run("Get last segment", func(t *testing.T) {
defer DeleteAll{}.Check(ctx, t, db)
createObject(ctx, t, db, obj, 2)
expectedSegmentSecond := metabase.Segment{
StreamID: obj.StreamID,
Position: metabase.SegmentPosition{
Index: 2,
},
RootPieceID: storj.PieceID{1},
EncryptedKey: []byte{3},
EncryptedKeyNonce: []byte{4},
EncryptedSize: 1024,
PlainSize: 512,
Pieces: metabase.Pieces{{Number: 0, StorageNode: storj.NodeID{2}}},
Redundancy: defaultTestRedundancy,
}
expectedSegmentFirst := expectedSegmentSecond
expectedSegmentFirst.Position.Index = 1
GetLatestObjectLastSegment{
Opts: metabase.GetLatestObjectLastSegment{
ObjectLocation: location,
},
Result: expectedSegmentSecond,
}.Check(ctx, t, db)
Verify{
Objects: []metabase.RawObject{
{
ObjectStream: obj,
CreatedAt: now,
Status: metabase.Committed,
Encryption: defaultTestEncryption,
},
},
Segments: []metabase.RawSegment{
metabase.RawSegment(expectedSegmentFirst),
metabase.RawSegment(expectedSegmentSecond),
},
}.Check(ctx, t, db)
})
})
}

View File

@ -148,6 +148,21 @@ func (step GetSegmentByPosition) Check(ctx *testcontext.Context, t *testing.T, d
require.Zero(t, diff) require.Zero(t, diff)
} }
type GetLatestObjectLastSegment struct {
Opts metabase.GetLatestObjectLastSegment
Result metabase.Segment
ErrClass *errs.Class
ErrText string
}
func (step GetLatestObjectLastSegment) Check(ctx *testcontext.Context, t *testing.T, db *metabase.DB) {
result, err := db.GetLatestObjectLastSegment(ctx, step.Opts)
checkError(t, err, step.ErrClass, step.ErrText)
diff := cmp.Diff(step.Result, result, cmpopts.EquateApproxTime(5*time.Second))
require.Zero(t, diff)
}
type DeleteObjectExactVersion struct { type DeleteObjectExactVersion struct {
Opts metabase.DeleteObjectExactVersion Opts metabase.DeleteObjectExactVersion
Result metabase.DeleteObjectResult Result metabase.DeleteObjectResult