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

satellite/metabase: copy inside transaction

Browse Source 
Read the source object and write the destination object in the same
transaction, to prevent breaking the object because it was deleted
simultaneously.

This is probably the root cause of the metainfo loop halting from
2022-06-21 onwards, where 2 objects lost their root_piece_id during
copying.

Part of https://github.com/storj/storj/issues/4930

Change-Id: I9c45d56a7bfb48ecd5f4906ee1cca42922901e90
This commit is contained in:
Erik van Velzen 2022-06-24 01:15:42 +02:00
parent e3ac0ae2c2
commit 77fea6137f
1 changed files with 96 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

189
satellite/metabase/copy_object.go
View File

@ -162,76 +162,79 @@ func (db *DB) FinishCopyObject(ctx context.Context, opts FinishCopyObject) (obje
} }
originalObject := Object{} originalObject := Object{}
copyObject := Object{}
var copyMetadata []byte
var ancestorStreamIDBytes []byte var ancestorStreamIDBytes []byte
err = db.db.QueryRowContext(ctx, ` err = txutil.WithTx(ctx, db.db, nil, func(ctx context.Context, tx tagsql.Tx) (err error) {
SELECT err = tx.QueryRowContext(ctx, `
objects.stream_id, SELECT
expires_at, objects.stream_id,
segment_count, expires_at,
encrypted_metadata, segment_count,
total_plain_size, total_encrypted_size, fixed_segment_size, encrypted_metadata,
encryption, total_plain_size, total_encrypted_size, fixed_segment_size,
segment_copies.ancestor_stream_id encryption,
FROM objects segment_copies.ancestor_stream_id
LEFT JOIN segment_copies ON objects.stream_id = segment_copies.stream_id FROM objects
WHERE LEFT JOIN segment_copies ON objects.stream_id = segment_copies.stream_id
project_id = $1 AND WHERE
bucket_name = $2 AND project_id = $1 AND
object_key = $3 AND bucket_name = $2 AND
version = $4 AND object_key = $3 AND
status = `+committedStatus, version = $4 AND
opts.ProjectID, []byte(opts.BucketName), opts.ObjectKey, opts.Version). status = `+committedStatus,
Scan( opts.ProjectID, []byte(opts.BucketName), opts.ObjectKey, opts.Version).
&originalObject.StreamID, Scan(
&originalObject.ExpiresAt, &originalObject.StreamID,
&originalObject.SegmentCount, &originalObject.ExpiresAt,
&originalObject.EncryptedMetadata, &originalObject.SegmentCount,
&originalObject.TotalPlainSize, &originalObject.TotalEncryptedSize, &originalObject.FixedSegmentSize, &originalObject.EncryptedMetadata,
encryptionParameters{&originalObject.Encryption}, &originalObject.TotalPlainSize, &originalObject.TotalEncryptedSize, &originalObject.FixedSegmentSize,
&ancestorStreamIDBytes, encryptionParameters{&originalObject.Encryption},
) &ancestorStreamIDBytes,
if err != nil { )
if errors.Is(err, sql.ErrNoRows) { if err != nil {
return Object{}, storj.ErrObjectNotFound.Wrap(Error.Wrap(err)) if errors.Is(err, sql.ErrNoRows) {
return storj.ErrObjectNotFound.Wrap(Error.Wrap(err))
}
return Error.New("unable to query object status: %w", err)
} }
return Object{}, Error.New("unable to query object status: %w", err) originalObject.BucketName = opts.BucketName
} originalObject.ProjectID = opts.ProjectID
originalObject.BucketName = opts.BucketName originalObject.Version = opts.Version
originalObject.ProjectID = opts.ProjectID originalObject.Status = Committed
originalObject.Version = opts.Version
originalObject.Status = Committed
if int(originalObject.SegmentCount) != len(opts.NewSegmentKeys) { if int(originalObject.SegmentCount) != len(opts.NewSegmentKeys) {
return Object{}, ErrInvalidRequest.New("wrong amount of segments keys received (received %d, need %d)", originalObject.SegmentCount, len(opts.NewSegmentKeys)) return ErrInvalidRequest.New("wrong amount of segments keys received (received %d, need %d)", originalObject.SegmentCount, len(opts.NewSegmentKeys))
} }
var newSegments struct { var newSegments struct {
Positions []int64 Positions []int64
EncryptedKeys [][]byte EncryptedKeys [][]byte
EncryptedKeyNonces [][]byte EncryptedKeyNonces [][]byte
} }
for _, u := range opts.NewSegmentKeys { for _, u := range opts.NewSegmentKeys {
newSegments.EncryptedKeys = append(newSegments.EncryptedKeys, u.EncryptedKey) newSegments.EncryptedKeys = append(newSegments.EncryptedKeys, u.EncryptedKey)
newSegments.EncryptedKeyNonces = append(newSegments.EncryptedKeyNonces, u.EncryptedKeyNonce) newSegments.EncryptedKeyNonces = append(newSegments.EncryptedKeyNonces, u.EncryptedKeyNonce)
newSegments.Positions = append(newSegments.Positions, int64(u.Position.Encode())) newSegments.Positions = append(newSegments.Positions, int64(u.Position.Encode()))
} }
positions := make([]int64, originalObject.SegmentCount) positions := make([]int64, originalObject.SegmentCount)
rootPieceIDs := make([][]byte, originalObject.SegmentCount) rootPieceIDs := make([][]byte, originalObject.SegmentCount)
expiresAts := make([]*time.Time, originalObject.SegmentCount) expiresAts := make([]*time.Time, originalObject.SegmentCount)
encryptedSizes := make([]int32, originalObject.SegmentCount) encryptedSizes := make([]int32, originalObject.SegmentCount)
plainSizes := make([]int32, originalObject.SegmentCount) plainSizes := make([]int32, originalObject.SegmentCount)
plainOffsets := make([]int64, originalObject.SegmentCount) plainOffsets := make([]int64, originalObject.SegmentCount)
inlineDatas := make([][]byte, originalObject.SegmentCount) inlineDatas := make([][]byte, originalObject.SegmentCount)
redundancySchemes := make([]int64, originalObject.SegmentCount) redundancySchemes := make([]int64, originalObject.SegmentCount)
// TODO: there are probably columns that we can skip // TODO: there are probably columns that we can skip
// maybe it's possible to have the select and the insert in one query // maybe it's possible to have the select and the insert in one query
err = withRows(db.db.QueryContext(ctx, ` err = withRows(tx.QueryContext(ctx, `
SELECT SELECT
position, position,
expires_at, expires_at,
@ -244,48 +247,48 @@ func (db *DB) FinishCopyObject(ctx context.Context, opts FinishCopyObject) (obje
ORDER BY position ASC ORDER BY position ASC
LIMIT $2 LIMIT $2
`, originalObject.StreamID, originalObject.SegmentCount))(func(rows tagsql.Rows) error { `, originalObject.StreamID, originalObject.SegmentCount))(func(rows tagsql.Rows) error {
index := 0 index := 0
for rows.Next() { for rows.Next() {
err = rows.Scan( err = rows.Scan(
&positions[index], &positions[index],
&expiresAts[index], &expiresAts[index],
&rootPieceIDs[index], &rootPieceIDs[index],
&encryptedSizes[index], &plainOffsets[index], &plainSizes[index], &encryptedSizes[index], &plainOffsets[index], &plainSizes[index],
&redundancySchemes[index], &redundancySchemes[index],
&inlineDatas[index], &inlineDatas[index],
) )
if err != nil { if err != nil {
return err
}
index++
}
if err = rows.Err(); err != nil {
return err return err
} }
index++ return nil
})
if err != nil {
return Error.New("unable to copy object: %w", err)
} }
if err = rows.Err(); err != nil { onlyInlineSegments := true
return err for index := range positions {
if newSegments.Positions[index] != positions[index] {
return Error.New("missing new segment keys for segment %d", positions[index])
}
if onlyInlineSegments && (encryptedSizes[index] > 0) && len(inlineDatas[index]) == 0 {
onlyInlineSegments = false
}
} }
return nil
})
if err != nil {
return Object{}, Error.New("unable to copy object: %w", err)
}
onlyInlineSegments := true copyMetadata = originalObject.EncryptedMetadata
for index := range positions { if opts.OverrideMetadata {
if newSegments.Positions[index] != positions[index] { copyMetadata = opts.NewEncryptedMetadata
return Object{}, Error.New("missing new segment keys for segment %d", positions[index])
} }
if onlyInlineSegments && (encryptedSizes[index] > 0) && len(inlineDatas[index]) == 0 {
onlyInlineSegments = false
}
}
copyMetadata := originalObject.EncryptedMetadata copyObject = originalObject
if opts.OverrideMetadata {
copyMetadata = opts.NewEncryptedMetadata
}
copyObject := originalObject
err = txutil.WithTx(ctx, db.db, nil, func(ctx context.Context, tx tagsql.Tx) (err error) {
// TODO we need to handle metadata correctly (copy from original object or replace) // TODO we need to handle metadata correctly (copy from original object or replace)
row := tx.QueryRowContext(ctx, ` row := tx.QueryRowContext(ctx, `
WITH existing_object AS ( WITH existing_object AS (