Fix SQL driver transaction commit handling in Create and Delete

Both Create and Delete committed their transaction via
'defer transaction.Commit()' with the error discarded, which breaks in
two ways:

- If COMMIT fails (connection drop, server-side error), Create returns
  nil while the release record was never persisted, and Delete returns
  the release as if deleted while nothing was deleted. Release history
  is silently lost either way.
- In Delete the defer was registered before the DELETE statements ran,
  so error paths after it (labels lookup failure, labels delete
  failure) still committed a partial deletion while reporting failure
  to the caller.

Commit explicitly at the end of each function, check the error, and
roll back on the intermediate error paths in Delete. Also fix the
custom_labels down migration which used 'DELETE TABLE' (not valid SQL)
instead of 'DROP TABLE'.

Regression tests cover COMMIT failure for both Create and Delete via
sqlmock.

Signed-off-by: Mukul <nmukul32@gmail.com>

pkg/storage/driver/sql.go

@@ -231,7 +231,7 @@ func (s *SQL) ensureDBSetup() error {
 				},
 				Down: []string{
 					fmt.Sprintf(`
-						DELETE TABLE %s;
+						DROP TABLE %s;
 					`, sqlCustomLabelsTableName),
 				},
 			},
@@ -569,7 +569,11 @@ func (s *SQL) Create(key string, rel release.Releaser) error {
 			return err
 		}
 	}
-	defer transaction.Commit()
+
+	if err := transaction.Commit(); err != nil {
+		s.Logger().Debug("failed to commit release creation", slog.String("key", key), slog.Any("error", err))
+		return fmt.Errorf("failed to commit release creation: %w", err)
+	}
 
 	return nil
 }
@@ -649,7 +653,6 @@ func (s *SQL) Delete(key string) (release.Releaser, error) {
 		transaction.Rollback()
 		return nil, err
 	}
-	defer transaction.Commit()
 
 	deleteQuery, args, err := s.statementBuilder.
 		Delete(sqlReleaseTableName).
@@ -658,12 +661,14 @@ func (s *SQL) Delete(key string) (release.Releaser, error) {
 		ToSql()
 	if err != nil {
 		s.Logger().Debug("failed to build delete query", slog.Any("error", err))
+		transaction.Rollback()
 		return nil, err
 	}
 
 	_, err = transaction.Exec(deleteQuery, args...)
 	if err != nil {
 		s.Logger().Debug("failed perform delete query", slog.Any("error", err))
+		transaction.Rollback()
 		return release, err
 	}
 
@@ -673,6 +678,7 @@ func (s *SQL) Delete(key string) (release.Releaser, error) {
 			slog.String("namespace", s.namespace),
 			slog.String("key", key),
 			slog.Any("error", err))
+		transaction.Rollback()
 		return nil, err
 	}
 
@@ -684,10 +690,21 @@ func (s *SQL) Delete(key string) (release.Releaser, error) {
 
 	if err != nil {
 		s.Logger().Debug("failed to build delete Labels query", slog.Any("error", err))
+		transaction.Rollback()
 		return nil, err
 	}
-	_, err = transaction.Exec(deleteCustomLabelsQuery, args...)
-	return release, err
+
+	if _, err = transaction.Exec(deleteCustomLabelsQuery, args...); err != nil {
+		transaction.Rollback()
+		return release, err
+	}
+
+	if err := transaction.Commit(); err != nil {
+		s.Logger().Debug("failed to commit release deletion", slog.String("key", key), slog.Any("error", err))
+		return nil, fmt.Errorf("failed to commit release deletion: %w", err)
+	}
+
+	return release, nil
 }
 
 // Get release custom labels from database

pkg/storage/driver/sql_test.go

@@ -263,6 +263,66 @@ func TestSqlCreate(t *testing.T) {
 	}
 }
 
+// A COMMIT failure must surface as an error: previously the commit ran in a
+// defer with its error discarded, so Create reported success while the
+// release record was never persisted.
+func TestSqlCreateCommitFailure(t *testing.T) {
+	vers := 1
+	name := "smug-pigeon"
+	namespace := "default"
+	key := testKey(name, vers)
+	rel := releaseStub(name, vers, namespace, common.StatusDeployed)
+
+	sqlDriver, mock := newTestFixtureSQL(t)
+	body, _ := encodeRelease(rel)
+
+	query := fmt.Sprintf(
+		"INSERT INTO %s (%s,%s,%s,%s,%s,%s,%s,%s,%s) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)",
+		sqlReleaseTableName,
+		sqlReleaseTableKeyColumn,
+		sqlReleaseTableTypeColumn,
+		sqlReleaseTableBodyColumn,
+		sqlReleaseTableNameColumn,
+		sqlReleaseTableNamespaceColumn,
+		sqlReleaseTableVersionColumn,
+		sqlReleaseTableStatusColumn,
+		sqlReleaseTableOwnerColumn,
+		sqlReleaseTableCreatedAtColumn,
+	)
+
+	mock.ExpectBegin()
+	mock.
+		ExpectExec(regexp.QuoteMeta(query)).
+		WithArgs(key, sqlReleaseDefaultType, body, rel.Name, rel.Namespace, int(rel.Version), rel.Info.Status.String(), sqlReleaseDefaultOwner, recentUnixTimestamp()).
+		WillReturnResult(sqlmock.NewResult(1, 1))
+
+	labelsQuery := fmt.Sprintf(
+		"INSERT INTO %s (%s,%s,%s,%s) VALUES ($1,$2,$3,$4)",
+		sqlCustomLabelsTableName,
+		sqlCustomLabelsTableReleaseKeyColumn,
+		sqlCustomLabelsTableReleaseNamespaceColumn,
+		sqlCustomLabelsTableKeyColumn,
+		sqlCustomLabelsTableValueColumn,
+	)
+
+	mock.MatchExpectationsInOrder(false)
+	for k, v := range filterSystemLabels(rel.Labels) {
+		mock.
+			ExpectExec(regexp.QuoteMeta(labelsQuery)).
+			WithArgs(key, rel.Namespace, k, v).
+			WillReturnResult(sqlmock.NewResult(1, 1))
+	}
+	mock.ExpectCommit().WillReturnError(errors.New("connection reset by peer"))
+
+	if err := sqlDriver.Create(key, rel); err == nil {
+		t.Fatal("expected Create to fail when COMMIT fails, got nil")
+	}
+
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("sql expectations weren't met: %v", err)
+	}
+}
+
 func TestSqlCreateAlreadyExists(t *testing.T) {
 	vers := 1
 	name := "smug-pigeon"
@@ -556,6 +616,75 @@ func TestSqlDelete(t *testing.T) {
 	}
 }
 
+// A COMMIT failure must surface as an error: previously the commit ran in a
+// defer registered before the DELETE statements, so any error after it still
+// committed a partial deletion while Delete reported failure — or, on commit
+// failure, reported success while nothing was deleted.
+func TestSqlDeleteCommitFailure(t *testing.T) {
+	vers := 1
+	name := "smug-pigeon"
+	namespace := "default"
+	key := testKey(name, vers)
+	rel := releaseStub(name, vers, namespace, common.StatusDeployed)
+	body, _ := encodeRelease(rel)
+
+	sqlDriver, mock := newTestFixtureSQL(t)
+
+	selectQuery := fmt.Sprintf(
+		"SELECT %s FROM %s WHERE %s = $1 AND %s = $2",
+		sqlReleaseTableBodyColumn,
+		sqlReleaseTableName,
+		sqlReleaseTableKeyColumn,
+		sqlReleaseTableNamespaceColumn,
+	)
+
+	mock.ExpectBegin()
+	mock.
+		ExpectQuery(regexp.QuoteMeta(selectQuery)).
+		WithArgs(key, namespace).
+		WillReturnRows(
+			mock.NewRows([]string{
+				sqlReleaseTableBodyColumn,
+			}).AddRow(
+				body,
+			),
+		).RowsWillBeClosed()
+
+	deleteQuery := fmt.Sprintf(
+		"DELETE FROM %s WHERE %s = $1 AND %s = $2",
+		sqlReleaseTableName,
+		sqlReleaseTableKeyColumn,
+		sqlReleaseTableNamespaceColumn,
+	)
+	mock.
+		ExpectExec(regexp.QuoteMeta(deleteQuery)).
+		WithArgs(key, namespace).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+
+	mockGetReleaseCustomLabels(mock, key, namespace, rel.Labels)
+
+	deleteLabelsQuery := fmt.Sprintf(
+		"DELETE FROM %s WHERE %s = $1 AND %s = $2",
+		sqlCustomLabelsTableName,
+		sqlCustomLabelsTableReleaseKeyColumn,
+		sqlCustomLabelsTableReleaseNamespaceColumn,
+	)
+	mock.
+		ExpectExec(regexp.QuoteMeta(deleteLabelsQuery)).
+		WithArgs(key, namespace).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+
+	mock.ExpectCommit().WillReturnError(errors.New("connection reset by peer"))
+
+	if _, err := sqlDriver.Delete(key); err == nil {
+		t.Fatal("expected Delete to fail when COMMIT fails, got nil")
+	}
+
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("sql expectations weren't met: %v", err)
+	}
+}
+
 func mockGetReleaseCustomLabels(mock sqlmock.Sqlmock, key string, namespace string, labels map[string]string) {
 	query := fmt.Sprintf(
 		regexp.QuoteMeta("SELECT %s, %s FROM %s WHERE %s = $1 AND %s = $2"),