satellite/metabase: remove old object segments on overwrite

While adding support for pending_objects table one case was missed.
When we are uploading object to location where old objects exists
we are not removing old object segments at all. Old object is
overwritten with new object metadata but segments remains without
corresponding object. This fix removes all existing committed objects
(with it's segments) before committing new object.

https://github.com/storj/storj/issues/6255

Change-Id: Id657840edf763fd6aec8191788d819191b074fb7

satellite/metabase/commit.go

@@ -12,7 +12,6 @@ import (
 	pgxerrcode "github.com/jackc/pgerrcode"
 	"github.com/zeebo/errs"
 	"go.uber.org/zap"
-	"golang.org/x/exp/slices"
 
 	"storj.io/common/memory"
 	"storj.io/common/storj"
@@ -729,14 +728,25 @@ func (db *DB) CommitObject(ctx context.Context, opts CommitObject) (object Objec
 		}
 
 		if opts.UsePendingObjectsTable {
-			opts.Version = 1
-
-			// we remove from deletion list object with version 1 as we will
-			// override/update instead deleting and inserting new one.
-			index := slices.Index(versionsToDelete, opts.Version)
-			if index != -1 {
-				versionsToDelete = slices.Delete(versionsToDelete, index, index+1)
+			// remove existing object(s) before inserting new one
+			// TODO after switching to pending_objects table completely we should
+			// be able to just delete all objects under this key and avoid
+			// selecting versions from above
+			for _, version := range versionsToDelete {
+				_, err := db.deleteObjectExactVersion(ctx, DeleteObjectExactVersion{
+					ObjectLocation: ObjectLocation{
+						ProjectID:  opts.ProjectID,
+						BucketName: opts.BucketName,
+						ObjectKey:  opts.ObjectKey,
+					},
+					Version: version,
+				}, tx)
+				if err != nil {
+					return Error.New("failed to delete existing object: %w", err)
 				}
+			}
+
+			opts.Version = 1
 
 			args = append(args,
 				opts.EncryptedMetadataNonce,
@@ -792,18 +802,8 @@ func (db *DB) CommitObject(ctx context.Context, opts CommitObject) (object Objec
 					encrypted_metadata_nonce, encrypted_metadata, encrypted_metadata_encrypted_key
 				)
 				SELECT * FROM object_to_commit
-				ON CONFLICT (project_id, bucket_name, object_key, version)
-				DO UPDATE SET (
-					stream_id,
-					status, segment_count, total_plain_size, total_encrypted_size, fixed_segment_size, zombie_deletion_deadline,
-					encryption,
-					encrypted_metadata_nonce, encrypted_metadata, encrypted_metadata_encrypted_key
-				) = (SELECT
-						stream_id,
-						status, segment_count, total_plain_size, total_encrypted_size, fixed_segment_size, zombie_deletion_deadline,
-						encryption,
-						encrypted_metadata_nonce, encrypted_metadata, encrypted_metadata_encrypted_key
-					FROM object_to_commit)
+				-- we don't want ON CONFLICT clause to update existign object
+				-- as this way we may miss removing old object segments
 				RETURNING
 					created_at, expires_at,
 					encrypted_metadata, encrypted_metadata_encrypted_key, encrypted_metadata_nonce,
@@ -813,6 +813,15 @@ func (db *DB) CommitObject(ctx context.Context, opts CommitObject) (object Objec
 				&object.EncryptedMetadata, &object.EncryptedMetadataEncryptedKey, &object.EncryptedMetadataNonce,
 				encryptionParameters{&object.Encryption},
 			)
+			if err != nil {
+				if errors.Is(err, sql.ErrNoRows) {
+					return ErrObjectNotFound.Wrap(Error.New("object with specified version and pending status is missing"))
+				} else if code := pgerrcode.FromError(err); code == pgxerrcode.NotNullViolation {
+					// TODO maybe we should check message if 'encryption' label is there
+					return ErrInvalidRequest.New("Encryption is missing")
+				}
+				return Error.New("failed to update object: %w", err)
+			}
 		} else {
 			metadataColumns := ""
 			if opts.OverrideEncryptedMetadata {
@@ -860,7 +869,6 @@ func (db *DB) CommitObject(ctx context.Context, opts CommitObject) (object Objec
 				&object.EncryptedMetadata, &object.EncryptedMetadataEncryptedKey, &object.EncryptedMetadataNonce,
 				encryptionParameters{&object.Encryption},
 			)
-		}
 			if err != nil {
 				if errors.Is(err, sql.ErrNoRows) {
 					return ErrObjectNotFound.Wrap(Error.New("object with specified version and pending status is missing"))
@@ -884,6 +892,7 @@ func (db *DB) CommitObject(ctx context.Context, opts CommitObject) (object Objec
 					return Error.New("failed to delete existing object: %w", err)
 				}
 			}
+		}
 
 		object.StreamID = opts.StreamID
 		object.ProjectID = opts.ProjectID

satellite/metabase/commit_test.go

@@ -4285,6 +4285,19 @@ func TestCommitObject(t *testing.T) {
 					}.Check(ctx, t, db)
 					obj.Version++
 
+					expectedInlineData := testrand.Bytes(8)
+					expectedEncryptedKey := testrand.Bytes(32)
+					expectedEncryptedKeyNonce := testrand.Bytes(32)
+					metabasetest.CommitInlineSegment{
+						Opts: metabase.CommitInlineSegment{
+							ObjectStream:           obj,
+							InlineData:             expectedInlineData,
+							EncryptedKey:           expectedEncryptedKey,
+							EncryptedKeyNonce:      expectedEncryptedKeyNonce,
+							UsePendingObjectsTable: true,
+						},
+					}.Check(ctx, t, db)
+
 					metabasetest.CommitObject{
 						Opts: metabase.CommitObject{
 							ObjectStream: obj,
@@ -4304,10 +4317,22 @@ func TestCommitObject(t *testing.T) {
 								ObjectStream:       obj,
 								CreatedAt:          now,
 								Status:             metabase.Committed,
+								SegmentCount:       1,
+								TotalEncryptedSize: int64(len(expectedInlineData)),
 
 								Encryption: metabasetest.DefaultEncryption,
 							},
 						},
+						Segments: []metabase.RawSegment{
+							{
+								StreamID:          obj.StreamID,
+								EncryptedKey:      expectedEncryptedKey,
+								EncryptedKeyNonce: expectedEncryptedKeyNonce,
+								EncryptedSize:     int32(len(expectedInlineData)),
+								InlineData:        expectedInlineData,
+								CreatedAt:         time.Now(),
+							},
+						},
 					}.Check(ctx, t, db)
 				}
 			})

satellite/metainfo/endpoint_object_test.go

@@ -925,6 +925,19 @@ func TestEndpoint_Object_No_StorageNodes_UsePendingObjectsTable(t *testing.T) {
 				})
 			}
 		})
+
+		t.Run("override on upload with segments", func(t *testing.T) {
+			defer ctx.Check(deleteBucket)
+
+			for i := 0; i < 5; i++ {
+				err := planet.Uplinks[0].Upload(ctx, planet.Satellites[0], bucketName, "test-object", testrand.Bytes(1*memory.KiB))
+				require.NoError(t, err)
+			}
+
+			segments, err := planet.Satellites[0].Metabase.DB.TestingAllSegments(ctx)
+			require.NoError(t, err)
+			require.Len(t, segments, 1)
+		})
 	})
 }