JakeHillion/storj
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

satellite/{metabase, metainfo}: merge logic for BeginCopyObject and BeginMoveObject

Browse Source 
Closes https://github.com/storj/storj/issues/4514

Change-Id: I4d59fe9e3d30f67ce642d6266b878956e38f5db1
This commit is contained in:
Qweder93 2022-09-14 14:55:58 +03:00 committed by Storj Robot
parent 01735c82d8
commit c210776a36
3 changed files with 59 additions and 145 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
satellite/metabase/copy_object.go
View File

@ -5,8 +5,6 @@ package metabase
import (
"context"
"database/sql"
"errors"
"time"
"github.com/zeebo/errs"
@ -18,16 +16,8 @@ import (
"storj.io/private/tagsql"
)
// BeginCopyObjectResult holds data needed to finish copy object.
type BeginCopyObjectResult struct {
StreamID uuid.UUID
Version Version
EncryptedMetadata []byte
EncryptedMetadataKeyNonce []byte
EncryptedMetadataKey []byte
EncryptedKeysNonces []EncryptedKeyAndNonce
EncryptionParameters storj.EncryptionParameters
}
// BeginCopyObjectResult holds data needed to begin copy object.
type BeginCopyObjectResult BeginMoveCopyResults
// BeginCopyObject holds all data needed begin copy object method.
type BeginCopyObject struct {
@ -39,67 +29,13 @@ type BeginCopyObject struct {
}
// BeginCopyObject collects all data needed to begin object copy procedure.
func (db *DB) BeginCopyObject(ctx context.Context, opts BeginCopyObject) (result BeginCopyObjectResult, err error) {
defer mon.Task()(&ctx)(&err)
if err := opts.ObjectLocation.Verify(); err != nil {
return BeginCopyObjectResult{}, err
}
object, err := db.GetObjectLastCommitted(ctx, GetObjectLastCommitted{
ObjectLocation: ObjectLocation{
ProjectID: opts.ProjectID,
BucketName: opts.BucketName,
ObjectKey: opts.ObjectKey,
},
})
func (db *DB) BeginCopyObject(ctx context.Context, opts BeginCopyObject) (_ BeginCopyObjectResult, err error) {
result, err := db.beginMoveCopyObject(ctx, opts.ObjectLocation, CopySegmentLimit, opts.VerifyLimits)
if err != nil {
return BeginCopyObjectResult{}, err
}
if int64(object.SegmentCount) > CopySegmentLimit {
return BeginCopyObjectResult{}, ErrInvalidRequest.New("object to copy has too many segments (%d). Limit is %d.", object.SegmentCount, CopySegmentLimit)
}
if opts.VerifyLimits != nil {
err = opts.VerifyLimits(object.TotalEncryptedSize, int64(object.SegmentCount))
if err != nil {
return BeginCopyObjectResult{}, err
}
}
err = withRows(db.db.QueryContext(ctx, `
SELECT
position, encrypted_key_nonce, encrypted_key
FROM segments
WHERE stream_id = $1
ORDER BY stream_id, position ASC
`, object.StreamID))(func(rows tagsql.Rows) error {
for rows.Next() {
var keys EncryptedKeyAndNonce
err = rows.Scan(&keys.Position, &keys.EncryptedKeyNonce, &keys.EncryptedKey)
if err != nil {
return Error.New("failed to scan segments: %w", err)
}
result.EncryptedKeysNonces = append(result.EncryptedKeysNonces, keys)
}
return nil
})
if err != nil && !errors.Is(err, sql.ErrNoRows) {
return BeginCopyObjectResult{}, Error.New("unable to fetch object segments: %w", err)
}
result.StreamID = object.StreamID
result.Version = object.Version
result.EncryptionParameters = object.Encryption
result.EncryptedMetadata = object.EncryptedMetadata
result.EncryptedMetadataKey = object.EncryptedMetadataEncryptedKey
result.EncryptedMetadataKeyNonce = object.EncryptedMetadataNonce
return result, nil
return BeginCopyObjectResult(result), nil
}
// FinishCopyObject holds all data needed to finish object copy.

75
satellite/metabase/move_object.go
View File

@ -18,17 +18,8 @@ import (
"storj.io/private/tagsql"
)
// BeginMoveObjectResult holds data needed to finish move object.
type BeginMoveObjectResult struct {
StreamID uuid.UUID
Version Version
// TODO we need metadata because of an uplink issue with how we are storing key and nonce
EncryptedMetadata []byte
EncryptedMetadataKeyNonce []byte
EncryptedMetadataKey []byte
EncryptedKeysNonces []EncryptedKeyAndNonce
EncryptionParameters storj.EncryptionParameters
}
// BeginMoveObjectResult holds data needed to begin move object.
type BeginMoveObjectResult BeginMoveCopyResults
// EncryptedKeyAndNonce holds single segment position, encrypted key and nonce.
type EncryptedKeyAndNonce struct {
@ -42,27 +33,55 @@ type BeginMoveObject struct {
ObjectLocation
}
// BeginMoveCopyResults holds all data needed to begin move and copy object methods.
type BeginMoveCopyResults struct {
StreamID uuid.UUID
Version Version
EncryptedMetadata []byte
EncryptedMetadataKeyNonce []byte
EncryptedMetadataKey []byte
EncryptedKeysNonces []EncryptedKeyAndNonce
EncryptionParameters storj.EncryptionParameters
}
// BeginMoveObject collects all data needed to begin object move procedure.
func (db *DB) BeginMoveObject(ctx context.Context, opts BeginMoveObject) (result BeginMoveObjectResult, err error) {
defer mon.Task()(&ctx)(&err)
if err := opts.ObjectLocation.Verify(); err != nil {
return BeginMoveObjectResult{}, err
}
object, err := db.GetObjectLastCommitted(ctx, GetObjectLastCommitted{
ObjectLocation: ObjectLocation{
ProjectID: opts.ProjectID,
BucketName: opts.BucketName,
ObjectKey: opts.ObjectKey,
},
})
func (db *DB) BeginMoveObject(ctx context.Context, opts BeginMoveObject) (_ BeginMoveObjectResult, err error) {
result, err := db.beginMoveCopyObject(ctx, opts.ObjectLocation, MoveSegmentLimit, nil)
if err != nil {
return BeginMoveObjectResult{}, err
}
if int64(object.SegmentCount) > MoveSegmentLimit {
return BeginMoveObjectResult{}, ErrInvalidRequest.New("object to move has too many segments (%d). Limit is %d.", object.SegmentCount, MoveSegmentLimit)
return BeginMoveObjectResult(result), nil
}
// beginMoveCopyObject collects all data needed to begin object move/copy procedure.
func (db *DB) beginMoveCopyObject(ctx context.Context, location ObjectLocation, segmentLimit int64, verifyLimits func(encryptedObjectSize int64, nSegments int64) error) (result BeginMoveCopyResults, err error) {
defer mon.Task()(&ctx)(&err)
if err := location.Verify(); err != nil {
return BeginMoveCopyResults{}, err
}
object, err := db.GetObjectLastCommitted(ctx, GetObjectLastCommitted{
ObjectLocation: ObjectLocation{
ProjectID: location.ProjectID,
BucketName: location.BucketName,
ObjectKey: location.ObjectKey,
},
})
if err != nil {
return BeginMoveCopyResults{}, err
}
if int64(object.SegmentCount) > segmentLimit {
return BeginMoveCopyResults{}, ErrInvalidRequest.New("object has too many segments (%d). Limit is %d.", object.SegmentCount, CopySegmentLimit)
}
if verifyLimits != nil {
err = verifyLimits(object.TotalEncryptedSize, int64(object.SegmentCount))
if err != nil {
return BeginMoveCopyResults{}, err
}
}
err = withRows(db.db.QueryContext(ctx, `
@ -86,7 +105,7 @@ func (db *DB) BeginMoveObject(ctx context.Context, opts BeginMoveObject) (result
return nil
})
if err != nil && !errors.Is(err, sql.ErrNoRows) {
return BeginMoveObjectResult{}, Error.New("unable to fetch object segments: %w", err)
return BeginMoveCopyResults{}, Error.New("unable to fetch object segments: %w", err)
}
result.StreamID = object.StreamID

55
satellite/metainfo/endpoint_object.go
View File

@ -1785,57 +1785,16 @@ func (endpoint *Endpoint) BeginCopyObject(ctx context.Context, req *pb.ObjectBeg
}
func convertBeginCopyObjectResults(result metabase.BeginCopyObjectResult) (*pb.ObjectBeginCopyResponse, error) {
keys := make([]*pb.EncryptedKeyAndNonce, len(result.EncryptedKeysNonces))
for i, key := range result.EncryptedKeysNonces {
var nonce storj.Nonce
var err error
if len(key.EncryptedKeyNonce) != 0 {
nonce, err = storj.NonceFromBytes(key.EncryptedKeyNonce)
if err != nil {
return nil, err
}
}
keys[i] = &pb.EncryptedKeyAndNonce{
Position: &pb.SegmentPosition{
PartNumber: int32(key.Position.Part),
Index: int32(key.Position.Index),
},
EncryptedKey: key.EncryptedKey,
EncryptedKeyNonce: nonce,
}
}
// TODO we need this because of an uplink issue with how we are storing key and nonce
if result.EncryptedMetadataKey == nil {
streamMeta := &pb.StreamMeta{}
err := pb.Unmarshal(result.EncryptedMetadata, streamMeta)
if err != nil {
return nil, err
}
if streamMeta.LastSegmentMeta != nil {
result.EncryptedMetadataKey = streamMeta.LastSegmentMeta.EncryptedKey
result.EncryptedMetadataKeyNonce = streamMeta.LastSegmentMeta.KeyNonce
}
}
var metadataNonce storj.Nonce
var err error
if len(result.EncryptedMetadataKeyNonce) != 0 {
metadataNonce, err = storj.NonceFromBytes(result.EncryptedMetadataKeyNonce)
if err != nil {
return nil, err
}
beginMoveObjectResult, err := convertBeginMoveObjectResults(metabase.BeginMoveObjectResult(result))
if err != nil {
return nil, err
}
return &pb.ObjectBeginCopyResponse{
EncryptedMetadataKey: result.EncryptedMetadataKey,
EncryptedMetadataKeyNonce: metadataNonce,
EncryptionParameters: &pb.EncryptionParameters{
CipherSuite: pb.CipherSuite(result.EncryptionParameters.CipherSuite),
BlockSize: int64(result.EncryptionParameters.BlockSize),
},
SegmentKeys: keys,
EncryptedMetadataKeyNonce: beginMoveObjectResult.EncryptedMetadataKeyNonce,
EncryptedMetadataKey: beginMoveObjectResult.EncryptedMetadataKey,
SegmentKeys: beginMoveObjectResult.SegmentKeys,
EncryptionParameters: beginMoveObjectResult.EncryptionParameters,
}, nil
}