satellite/metabase: adjust BucketEmpty to use pending_objects table

Extends metabase.BucketEmpty logic to check also pending_objects
table for any entry.

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

Change-Id: Ia26c272de24a983b308a0b692e6bd5800487eb98

satellite/metabase/get.go

@@ -358,6 +358,13 @@ func (db *DB) BucketEmpty(ctx context.Context, opts BucketEmpty) (empty bool, er
 		SELECT
 			1
 		FROM objects
+		WHERE
+			project_id   = $1 AND
+			bucket_name  = $2
+		UNION ALL
+		SELECT
+			1
+		FROM pending_objects
 		WHERE
 			project_id   = $1 AND
 			bucket_name  = $2

satellite/metabase/get_test.go

@@ -1597,5 +1597,60 @@ func TestBucketEmpty(t *testing.T) {
 				},
 			}.Check(ctx, t, db)
 		})
+
+		t.Run("object in pending_objects", func(t *testing.T) {
+			defer metabasetest.DeleteAll{}.Check(ctx, t, db)
+
+			metabasetest.CreatePendingObjectNew(ctx, t, db, obj, 0)
+
+			metabasetest.BucketEmpty{
+				Opts: metabase.BucketEmpty{
+					ProjectID:  obj.ProjectID,
+					BucketName: obj.BucketName,
+				},
+				Result: false,
+			}.Check(ctx, t, db)
+
+			metabasetest.Verify{
+				PendingObjects: []metabase.RawPendingObject{
+					{
+						PendingObjectStream:    metabasetest.ObjectStreamToPending(obj),
+						CreatedAt:              now,
+						Encryption:             metabasetest.DefaultEncryption,
+						ZombieDeletionDeadline: &zombieDeadline,
+					},
+				},
+			}.Check(ctx, t, db)
+		})
+
+		t.Run("object in pending_objects and in object", func(t *testing.T) {
+			defer metabasetest.DeleteAll{}.Check(ctx, t, db)
+
+			object := metabasetest.CreateObject(ctx, t, db, obj, 0)
+
+			metabasetest.CreatePendingObjectNew(ctx, t, db, obj, 0)
+
+			metabasetest.BucketEmpty{
+				Opts: metabase.BucketEmpty{
+					ProjectID:  obj.ProjectID,
+					BucketName: obj.BucketName,
+				},
+				Result: false,
+			}.Check(ctx, t, db)
+
+			metabasetest.Verify{
+				Objects: []metabase.RawObject{
+					metabase.RawObject(object),
+				},
+				PendingObjects: []metabase.RawPendingObject{
+					{
+						PendingObjectStream:    metabasetest.ObjectStreamToPending(obj),
+						CreatedAt:              now,
+						Encryption:             metabasetest.DefaultEncryption,
+						ZombieDeletionDeadline: &zombieDeadline,
+					},
+				},
+			}.Check(ctx, t, db)
+		})
 	})
 }