stores: call pruneSlabs every time an object or multipart object were…

… deleted

stores/metadata.go

@@ -2685,20 +2685,32 @@ func archiveContracts(ctx context.Context, tx *gorm.DB, contracts []dbContract,
 	return nil
 }
 
+func pruneSlabs(tx *gorm.DB) error {
+	// delete slabs without any associated slices or buffers
+	return tx.Exec(`
+DELETE
+FROM slabs sla
+WHERE NOT EXISTS (SELECT 1 FROM slices sli WHERE sli.db_slab_id = sla.id)
+AND sla.db_buffered_slab_id IS NULL
+`).Error
+}
+
 // deleteObject deletes an object from the store and prunes all slabs which are
 // without an obect after the deletion. That means in case of packed uploads,
 // the slab is only deleted when no more objects point to it.
-func (s *SQLStore) deleteObject(tx *gorm.DB, bucket string, path string) (numDeleted int64, _ error) {
+func (s *SQLStore) deleteObject(tx *gorm.DB, bucket string, path string) (int64, error) {
 	tx = tx.Where("object_id = ? AND ?", path, sqlWhereBucket("objects", bucket)).
 		Delete(&dbObject{})
 	if tx.Error != nil {
 		return 0, tx.Error
 	}
-	numDeleted = tx.RowsAffected
+	numDeleted := tx.RowsAffected
 	if numDeleted == 0 {
 		return 0, nil // nothing to prune if no object was deleted
+	} else if err := pruneSlabs(tx); err != nil {
+		return 0, err
 	}
-	return
+	return numDeleted, nil
 }
 
 // deleteObjects deletes a batch of objects from the database. The order of
@@ -2729,6 +2741,11 @@ func (s *SQLStore) deleteObjects(bucket string, path string) (numDeleted int64,
 			}
 			duration = time.Since(start)
 			rowsAffected = res.RowsAffected
+
+			// prune slabs if we deleted an object
+			if rowsAffected > 0 {
+				return pruneSlabs(tx)
+			}
 			return nil
 		}); err != nil {
 			return 0, fmt.Errorf("failed to delete objects: %w", err)

stores/multipart.go

@@ -295,6 +295,10 @@ func (s *SQLStore) AbortMultipartUpload(ctx context.Context, bucket, path string
 		if err != nil {
 			return fmt.Errorf("failed to delete multipart upload: %w", err)
 		}
+		// Prune the slabs.
+		if err := pruneSlabs(tx); err != nil {
+			return fmt.Errorf("failed to prune slabs: %w", err)
+		}
 		return nil
 	})
 }
@@ -435,6 +439,11 @@ func (s *SQLStore) CompleteMultipartUpload(ctx context.Context, bucket, path str
 		if err := tx.Delete(&mu).Error; err != nil {
 			return fmt.Errorf("failed to delete multipart upload: %w", err)
 		}
+
+		// Prune the slabs.
+		if err := pruneSlabs(tx); err != nil {
+			return fmt.Errorf("failed to prune slabs: %w", err)
+		}
 		return nil
 	})
 	if err != nil {