Merge e55162daaa into 62c7e845d1

pkg/storage/driver/sql.go

@@ -17,6 +17,9 @@ limitations under the License.
 package driver
 
 import (
+	"context"
+	"database/sql"
+	"errors"
 	"fmt"
 	"log/slog"
 	"maps"
@@ -318,10 +321,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)
@@ -518,8 +526,14 @@ func (s *SQL) Create(key string, rel release.Releaser) error {
 		return err
 	}
 
-	if _, err := transaction.Exec(insertQuery, args...); err != nil {
-		defer transaction.Rollback()
+	if _, err := transaction.ExecContext(context.Background(), insertQuery, args...); err != nil {
+		// 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).
@@ -529,17 +543,17 @@ func (s *SQL) Create(key string, rel release.Releaser) error {
 			ToSql()
 		if buildErr != nil {
 			s.Logger().Debug("failed to build select query", "error", buildErr)
-			return err
+			return fmt.Errorf("failed to create release %q: %w", key, err)
 		}
 
 		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
 		}
 
 		s.Logger().Debug("failed to store release in SQL database", slog.String("key", key), slog.Any("error", err))
-		return err
+		return fmt.Errorf("failed to create release %q: %w", key, err)
 	}
 
 	// Filtering labels before insert cause in SQL storage driver system releases are stored in separate columns of release table
@@ -564,7 +578,7 @@ func (s *SQL) Create(key string, rel release.Releaser) error {
 			return err
 		}
 
-		if _, err := transaction.Exec(insertLabelsQuery, args...); err != nil {
+		if _, err := transaction.ExecContext(context.Background(), insertLabelsQuery, args...); err != nil {
 			defer transaction.Rollback()
 			s.Logger().Debug("failed to write Labels", slog.Any("error", err))
 			return err
@@ -609,7 +623,7 @@ func (s *SQL) Update(key string, rel release.Releaser) error {
 		return err
 	}
 
-	if _, err := s.db.Exec(query, args...); err != nil {
+	if _, err := s.db.ExecContext(context.Background(), query, args...); err != nil {
 		s.Logger().Debug("failed to update release in SQL database", slog.String("key", key), slog.Any("error", err))
 		return err
 	}
@@ -639,8 +653,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)
@@ -661,7 +679,7 @@ func (s *SQL) Delete(key string) (release.Releaser, error) {
 		return nil, err
 	}
 
-	_, err = transaction.Exec(deleteQuery, args...)
+	_, err = transaction.ExecContext(context.Background(), deleteQuery, args...)
 	if err != nil {
 		s.Logger().Debug("failed perform delete query", slog.Any("error", err))
 		return release, err
@@ -685,7 +703,7 @@ func (s *SQL) Delete(key string) (release.Releaser, error) {
 		s.Logger().Debug("failed to build delete Labels query", slog.Any("error", err))
 		return nil, err
 	}
-	_, err = transaction.Exec(deleteCustomLabelsQuery, args...)
+	_, err = transaction.ExecContext(context.Background(), deleteCustomLabelsQuery, args...)
 	return release, err
 }
 

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,67 @@ func TestSqlCreateAlreadyExists(t *testing.T) {
 				key,
 			),
 		).RowsWillBeClosed()
+
+	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")
+}
+
+func TestSqlCreateInsertFailureNotAlreadyExists(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)
+
+	insertQuery := 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,
+	)
+
+	// The insert fails for a reason unrelated to a duplicate key, e.g. the
+	// database is unreachable.
+	insertErr := errors.New("connection refused")
+	mock.ExpectBegin()
+	mock.
+		ExpectExec(regexp.QuoteMeta(insertQuery)).
+		WithArgs(key, sqlReleaseDefaultType, body, rel.Name, rel.Namespace, int(rel.Version), rel.Info.Status.String(), sqlReleaseDefaultOwner, recentUnixTimestamp()).
+		WillReturnError(insertErr)
+
+	selectQuery := fmt.Sprintf(
+		regexp.QuoteMeta("SELECT %s FROM %s WHERE %s = $1 AND %s = $2"),
+		sqlReleaseTableKeyColumn,
+		sqlReleaseTableName,
+		sqlReleaseTableKeyColumn,
+		sqlReleaseTableNamespaceColumn,
+	)
+
 	mock.ExpectRollback()
 
-	require.Errorf(t, sqlDriver.Create(key, rel), "failed to create release with key %s", key)
+	// No row comes back, so the release does not already exist and the original
+	// insert error has to be surfaced instead of ErrReleaseExists.
+	mock.
+		ExpectQuery(selectQuery).
+		WithArgs(key, namespace).
+		WillReturnError(sql.ErrNoRows)
+
+	err := sqlDriver.Create(key, rel)
+	require.Errorf(t, err, "expected Create to fail when the insert fails with key %s", key)
+	require.NotErrorIs(t, err, ErrReleaseExists)
+	require.ErrorIs(t, err, insertErr)
 	assert.NoErrorf(t, mock.ExpectationsWereMet(), "sql expectations weren't met")
 }