fix(storage/sql): retry refresh token update on serialization failures

UpdateRefreshToken runs a read-modify-write inside a SERIALIZABLE
transaction. Under concurrent refresh-token rotation of the same token
(e.g. multiple kube clients refreshing at once), Postgres aborts the
conflicting transaction with SQLSTATE 40001 and the error surfaced to
clients as HTTP 500.

Add a bounded, jittered retry around the refresh-token update that
re-runs the transaction on transient serialization/deadlock failures,
detected per-driver (Postgres 40001/40P01, MySQL 1213/1205). Error
wraps in the SQL storage now use %w so the driver error can be matched
with errors.As. SQLite serializes writes and opts out (nil check).

Enables the previously-disabled refresh-token concurrency conformance
tests for Postgres and MySQL.

Refs: CPEC-711
Signed-off-by: Ronan <ronan.souza@wildlifestudios.com>

Author: RonnanSouza

storage/sql/config.go

@@ -4,6 +4,7 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"database/sql"
+	"errors"
 	"fmt"
 	"log/slog"
 	"net"
@@ -21,14 +22,18 @@ import (
 
 const (
 	// postgres error codes
-	pgErrUniqueViolation = "23505" // unique_violation
+	pgErrUniqueViolation      = "23505" // unique_violation
+	pgErrSerializationFailure = "40001" // serialization_failure
+	pgErrDeadlockDetected     = "40P01" // deadlock_detected
 )
 
 const (
 	// MySQL error codes
 	mysqlErrDupEntry            = 1062
 	mysqlErrDupEntryWithKeyName = 1586
 	mysqlErrUnknownSysVar       = 1193
+	mysqlErrLockDeadlock        = 1213 // ER_LOCK_DEADLOCK
+	mysqlErrLockWaitTimeout     = 1205 // ER_LOCK_WAIT_TIMEOUT
 )
 
 const (
@@ -195,7 +200,15 @@ func (p *Postgres) open(logger *slog.Logger) (*conn, error) {
 		return sqlErr.Code == pgErrUniqueViolation
 	}
 
-	c := &conn{db, &flavorPostgres, logger, errCheck}
+	retryCheck := func(err error) bool {
+		var sqlErr *pq.Error
+		if !errors.As(err, &sqlErr) {
+			return false
+		}
+		return sqlErr.Code == pgErrSerializationFailure || sqlErr.Code == pgErrDeadlockDetected
+	}
+
+	c := &conn{db, &flavorPostgres, logger, errCheck, retryCheck}
 	if _, err := c.migrate(); err != nil {
 		return nil, fmt.Errorf("failed to perform migrations: %v", err)
 	}
@@ -307,7 +320,16 @@ func (s *MySQL) open(logger *slog.Logger) (*conn, error) {
 			sqlErr.Number == mysqlErrDupEntryWithKeyName
 	}
 
-	c := &conn{db, &flavorMySQL, logger, errCheck}
+	retryCheck := func(err error) bool {
+		var sqlErr *mysql.MySQLError
+		if !errors.As(err, &sqlErr) {
+			return false
+		}
+		return sqlErr.Number == mysqlErrLockDeadlock ||
+			sqlErr.Number == mysqlErrLockWaitTimeout
+	}
+
+	c := &conn{db, &flavorMySQL, logger, errCheck, retryCheck}
 	if _, err := c.migrate(); err != nil {
 		return nil, fmt.Errorf("failed to perform migrations: %v", err)
 	}

storage/sql/config_test.go

@@ -243,7 +243,7 @@ func TestPostgres(t *testing.T) {
 			Mode: pgSSLDisable, // Postgres container doesn't support SSL.
 		},
 	}
-	testDB(t, p, true, false)
+	testDB(t, p, true, true)
 }
 
 const testMySQLEnv = "DEX_MYSQL_HOST"
@@ -280,7 +280,7 @@ func TestMySQL(t *testing.T) {
 			"innodb_lock_wait_timeout": "3",
 		},
 	}
-	testDB(t, s, true, false)
+	testDB(t, s, true, true)
 }
 
 const testMySQL8Env = "DEX_MYSQL8_HOST"
@@ -317,5 +317,5 @@ func TestMySQL8(t *testing.T) {
 			"innodb_lock_wait_timeout": "3",
 		},
 	}
-	testDB(t, s, true, false)
+	testDB(t, s, true, true)
 }