Fix SQL driver error mapping for reads and duplicate creates

Two related error-contract problems in the SQL storage driver:

Get and Delete mapped *any* error from the release lookup to
ErrReleaseNotFound. A connection failure, a permission error or a
timeout was therefore indistinguishable from a release that genuinely
does not exist, and callers branch on that sentinel - a transient
database outage could be read as 'release absent'. Only sql.ErrNoRows
now maps to ErrReleaseNotFound; every other error is wrapped and
returned. Delete additionally left its transaction open on that path,
so it is now rolled back.

Create could never return ErrReleaseExists. After the insert fails the
transaction is in an aborted state, and PostgreSQL rejects every
subsequent statement on it, so the follow-up SELECT that decides
between 'already exists' and a genuine error always failed and the raw
driver error was surfaced instead. This diverges from the Secrets and
ConfigMaps drivers, which return ErrReleaseExists. The transaction is
now rolled back first and the existence check runs outside it.

Refs #32394 (items 2 and 3).

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

pkg/storage/driver/sql.go

@@ -17,6 +17,8 @@ limitations under the License.
 package driver
 
 import (
+	"database/sql"
+	"errors"
 	"fmt"
 	"log/slog"
 	"maps"
@@ -318,10 +320,15 @@ func (s *SQL) Get(key string) (release.Releaser, error) {
 		return nil, err
 	}
 
-	// Get will return an error if the result is empty
+	// Get will return sql.ErrNoRows if the result is empty. Any other error
+	// (connection failure, permission denied, timeout, ...) means we do not
+	// know whether the release exists, so it must not be reported as missing.
 	if err := s.db.Get(&record, query, args...); err != nil {
 		s.Logger().Debug("got SQL error when getting release", slog.String("key", key), slog.Any("error", err))
-		return nil, ErrReleaseNotFound
+		if errors.Is(err, sql.ErrNoRows) {
+			return nil, ErrReleaseNotFound
+		}
+		return nil, fmt.Errorf("failed to get release %q: %w", key, err)
 	}
 
 	release, err := decodeRelease(record.Body)
@@ -519,7 +526,13 @@ func (s *SQL) Create(key string, rel release.Releaser) error {
 	}
 
 	if _, err := transaction.Exec(insertQuery, args...); err != nil {
-		defer transaction.Rollback()
+		// The failed statement leaves the transaction in an aborted state -
+		// PostgreSQL rejects every subsequent statement on it with "current
+		// transaction is aborted" - so roll it back before checking whether the
+		// insert failed because the release already exists. Running that check
+		// on the transaction always errors out, which made ErrReleaseExists
+		// unreachable and surfaced the raw driver error instead.
+		transaction.Rollback()
 
 		selectQuery, args, buildErr := s.statementBuilder.
 			Select(sqlReleaseTableKeyColumn).
@@ -533,7 +546,7 @@ func (s *SQL) Create(key string, rel release.Releaser) error {
 		}
 
 		var record SQLReleaseWrapper
-		if err := transaction.Get(&record, selectQuery, args...); err == nil {
+		if getErr := s.db.Get(&record, selectQuery, args...); getErr == nil {
 			s.Logger().Debug("release already exists", slog.String("key", key))
 			return ErrReleaseExists
 		}
@@ -639,8 +652,12 @@ func (s *SQL) Delete(key string) (release.Releaser, error) {
 	var record SQLReleaseWrapper
 	err = transaction.Get(&record, selectQuery, args...)
 	if err != nil {
-		s.Logger().Debug("release not found", slog.String("key", key), slog.Any("error", err))
-		return nil, ErrReleaseNotFound
+		s.Logger().Debug("failed to get release for deletion", slog.String("key", key), slog.Any("error", err))
+		transaction.Rollback()
+		if errors.Is(err, sql.ErrNoRows) {
+			return nil, ErrReleaseNotFound
+		}
+		return nil, fmt.Errorf("failed to get release %q: %w", key, err)
 	}
 
 	release, err := decodeRelease(record.Body)

pkg/storage/driver/sql_test.go

@@ -14,6 +14,7 @@ limitations under the License.
 package driver
 
 import (
+	"database/sql"
 	"database/sql/driver"
 	"errors"
 	"fmt"
@@ -103,6 +104,97 @@ func TestSQLGet(t *testing.T) {
 	assert.NoErrorf(t, mock.ExpectationsWereMet(), "sql expectations weren't met")
 }
 
+// A real database failure must not be reported as "release not found": callers
+// branch on ErrReleaseNotFound and would treat an outage as an absent release.
+func TestSQLGetDatabaseError(t *testing.T) {
+	vers := int(1)
+	name := "smug-pigeon"
+	namespace := "default"
+	key := testKey(name, vers)
+
+	sqlDriver, mock := newTestFixtureSQL(t)
+
+	query := fmt.Sprintf(
+		regexp.QuoteMeta("SELECT %s FROM %s WHERE %s = $1 AND %s = $2"),
+		sqlReleaseTableBodyColumn,
+		sqlReleaseTableName,
+		sqlReleaseTableKeyColumn,
+		sqlReleaseTableNamespaceColumn,
+	)
+
+	dbErr := errors.New("connection refused")
+	mock.
+		ExpectQuery(query).
+		WithArgs(key, namespace).
+		WillReturnError(dbErr)
+
+	_, err := sqlDriver.Get(key)
+	require.Error(t, err, "expected an error when the database query fails")
+	require.NotErrorIs(t, err, ErrReleaseNotFound, "database failure must not be reported as ErrReleaseNotFound")
+	require.ErrorIs(t, err, dbErr, "expected the underlying database error to be wrapped")
+	assert.NoErrorf(t, mock.ExpectationsWereMet(), "sql expectations weren't met")
+}
+
+// An empty result set is the only case that means "release not found".
+func TestSQLGetNotFound(t *testing.T) {
+	vers := int(1)
+	name := "smug-pigeon"
+	namespace := "default"
+	key := testKey(name, vers)
+
+	sqlDriver, mock := newTestFixtureSQL(t)
+
+	query := fmt.Sprintf(
+		regexp.QuoteMeta("SELECT %s FROM %s WHERE %s = $1 AND %s = $2"),
+		sqlReleaseTableBodyColumn,
+		sqlReleaseTableName,
+		sqlReleaseTableKeyColumn,
+		sqlReleaseTableNamespaceColumn,
+	)
+
+	mock.
+		ExpectQuery(query).
+		WithArgs(key, namespace).
+		WillReturnError(sql.ErrNoRows)
+
+	_, err := sqlDriver.Get(key)
+	require.ErrorIs(t, err, ErrReleaseNotFound)
+	assert.NoErrorf(t, mock.ExpectationsWereMet(), "sql expectations weren't met")
+}
+
+// Same contract for Delete's existence check, which must also not leave the
+// transaction open when the lookup fails.
+func TestSQLDeleteDatabaseError(t *testing.T) {
+	vers := int(1)
+	name := "smug-pigeon"
+	namespace := "default"
+	key := testKey(name, vers)
+
+	sqlDriver, mock := newTestFixtureSQL(t)
+
+	selectQuery := fmt.Sprintf(
+		"SELECT %s FROM %s WHERE %s = $1 AND %s = $2",
+		sqlReleaseTableBodyColumn,
+		sqlReleaseTableName,
+		sqlReleaseTableKeyColumn,
+		sqlReleaseTableNamespaceColumn,
+	)
+
+	dbErr := errors.New("connection refused")
+	mock.ExpectBegin()
+	mock.
+		ExpectQuery(regexp.QuoteMeta(selectQuery)).
+		WithArgs(key, namespace).
+		WillReturnError(dbErr)
+	mock.ExpectRollback()
+
+	_, err := sqlDriver.Delete(key)
+	require.Error(t, err, "expected an error when the database query fails")
+	require.NotErrorIs(t, err, ErrReleaseNotFound, "database failure must not be reported as ErrReleaseNotFound")
+	require.ErrorIs(t, err, dbErr, "expected the underlying database error to be wrapped")
+	assert.NoErrorf(t, mock.ExpectationsWereMet(), "sql expectations weren't met")
+}
+
 func TestSQLList(t *testing.T) {
 	releases := []*rspb.Release{}
 	releases = append(releases,
@@ -269,6 +361,11 @@ func TestSqlCreateAlreadyExists(t *testing.T) {
 		sqlReleaseTableNamespaceColumn,
 	)
 
+	// The failed insert aborts the transaction, so it is rolled back before the
+	// existence check runs - the check has to happen outside the transaction or
+	// PostgreSQL rejects it and ErrReleaseExists can never be returned.
+	mock.ExpectRollback()
+
 	// Let's check that we do make sure the error is due to a release already existing
 	mock.
 		ExpectQuery(selectQuery).
@@ -280,9 +377,10 @@ func TestSqlCreateAlreadyExists(t *testing.T) {
 				key,
 			),
 		).RowsWillBeClosed()
-	mock.ExpectRollback()
 
-	require.Errorf(t, sqlDriver.Create(key, rel), "failed to create release with key %s", key)
+	err := sqlDriver.Create(key, rel)
+	require.Errorf(t, err, "expected Create to fail for an existing release with key %s", key)
+	require.ErrorIs(t, err, ErrReleaseExists)
 	assert.NoErrorf(t, mock.ExpectationsWereMet(), "sql expectations weren't met")
 }