stores: fix deletion of bucket with multipart uploads

bus/bus.go

@@ -311,7 +311,7 @@ func (b *bus) bucketHandlerDELETE(jc jape.Context) {
 	} else if name == "" {
 		jc.Error(errors.New("no name provided"), http.StatusBadRequest)
 		return
-	} else if jc.Check("failed to create bucket", b.ms.DeleteBucket(jc.Request.Context(), name)) != nil {
+	} else if jc.Check("failed to delete bucket", b.ms.DeleteBucket(jc.Request.Context(), name)) != nil {
 		return
 	}
 }

internal/testing/s3_test.go

@@ -373,9 +373,6 @@ func TestS3MultipartUploads(t *testing.T) {
 	core := cluster.S3Core
 	tt := cluster.tt
 
-	// delete default bucket before testing.
-	tt.OK(cluster.Bus.DeleteBucket(context.Background(), api.DefaultBucketName))
-
 	// Create bucket.
 	tt.OK(s3.MakeBucket(context.Background(), "multipart", minio.MakeBucketOptions{}))
 
@@ -386,15 +383,23 @@ func TestS3MultipartUploads(t *testing.T) {
 		t.Fatal("expected non-empty upload ID")
 	}
 
+	// Start another one in the default bucket. This should not show up when
+	// listing the uploads in the 'multipart' bucket.
+	tt.OKAll(core.NewMultipartUpload(context.Background(), api.DefaultBucketName, "foo", minio.PutObjectOptions{}))
+
 	// List uploads
 	lmu, err := core.ListMultipartUploads(context.Background(), "multipart", "", "", "", "", 0)
 	tt.OK(err)
 	if len(lmu.Uploads) != 1 {
-		t.Fatal("expected 1 upload")
+		t.Fatal("expected 1 upload", len(lmu.Uploads))
 	} else if upload := lmu.Uploads[0]; upload.UploadID != uploadID || upload.Key != "foo" {
 		t.Fatal("unexpected upload:", upload.UploadID, upload.Key)
 	}
 
+	// delete default bucket for the remainder of the test. This makes sure we
+	// can delete the bucket even though it contains a multipart upload.
+	tt.OK(cluster.Bus.DeleteBucket(context.Background(), api.DefaultBucketName))
+
 	// Add 3 parts out of order to make sure the object is reconstructed
 	// correctly.
 	putPart := func(partNum int, data []byte) string {

stores/metadata.go

@@ -518,7 +518,7 @@ func (s *SQLStore) DeleteBucket(ctx context.Context, bucket string) error {
 		if res.Error != nil {
 			return res.Error
 		}
-		return nil
+		return pruneSlabs(tx)
 	})
 }
 

stores/migrations.go

@@ -261,6 +261,12 @@ func performMigrations(db *gorm.DB, logger *zap.SugaredLogger) error {
 				return performMigration00020_missingIndices(tx, logger)
 			},
 		},
+		{
+			ID: "00021_multipoartUploadsBucketCascade",
+			Migrate: func(tx *gorm.DB) error {
+				return performMigration00021_multipartUploadsBucketCascade(tx, logger)
+			},
+		},
 	}
 	// Create migrator.
 	m := gormigrate.New(db, gormigrate.DefaultOptions, migrations)
@@ -938,3 +944,30 @@ func performMigration00020_missingIndices(txn *gorm.DB, logger *zap.SugaredLogge
 	logger.Info("migration 00020_missingIndices complete")
 	return nil
 }
+
+func performMigration00021_multipartUploadsBucketCascade(txn *gorm.DB, logger *zap.SugaredLogger) error {
+	logger.Info("performing migration 00021_multipoartUploadsBucketCascade")
+	// Disable foreign keys in SQLite to avoid issues with updating constraints.
+	if isSQLite(txn) {
+		if err := txn.Exec(`PRAGMA foreign_keys = 0`).Error; err != nil {
+			return err
+		}
+	}
+
+	// Add cascade constraint.
+	if err := txn.Migrator().AutoMigrate(&dbMultipartUpload{}); err != nil {
+		return err
+	}
+
+	// Enable foreign keys again.
+	if isSQLite(txn) {
+		if err := txn.Exec(`PRAGMA foreign_keys = 1`).Error; err != nil {
+			return err
+		}
+		if err := txn.Exec(`PRAGMA foreign_key_check(slices)`).Error; err != nil {
+			return err
+		}
+	}
+	logger.Info("migration 00021_multipoartUploadsBucketCascade complete")
+	return nil
+}

stores/multipart.go

@@ -22,9 +22,9 @@ type (
 		Model
 
 		Key        []byte
-		UploadID   string `gorm:"uniqueIndex;NOT NULL;size:64"`
-		ObjectID   string `gorm:"index;NOT NULL"`
-		DBBucket   dbBucket
+		UploadID   string            `gorm:"uniqueIndex;NOT NULL;size:64"`
+		ObjectID   string            `gorm:"index;NOT NULL"`
+		DBBucket   dbBucket          `gorm:"constraint:OnDelete:CASCADE"` // CASCADE to delete uploads when bucket is deleted
 		DBBucketID uint              `gorm:"index;NOT NULL"`
 		Parts      []dbMultipartPart `gorm:"constraint:OnDelete:CASCADE"` // CASCADE to delete parts too
 		MimeType   string            `gorm:"index"`
@@ -186,7 +186,7 @@ func (s *SQLStore) MultipartUploads(ctx context.Context, bucket, prefix, keyMark
 		err := tx.
 			Model(&dbMultipartUpload{}).
 			Joins("DBBucket").
-			Where("? AND ? AND ?", prefixExpr, keyMarkerExpr, uploadIDMarkerExpr).
+			Where("? AND ? AND ? AND DBBucket.name = ?", prefixExpr, keyMarkerExpr, uploadIDMarkerExpr, bucket).
 			Limit(limit).
 			Find(&dbUploads).
 			Error