[e2e-rca] Retry signing-key rotation on serialization failures (SQLSTATE 40001/40P01)

server/signer/local.go

@@ -62,7 +62,14 @@ func (l *localSigner) Start(ctx context.Context) {
 				return
 			case <-time.After(time.Second * 30):
 				if err := l.rotator.rotate(); err != nil {
-					l.logger.Error("failed to rotate keys", "err", err)
+					if err == errAlreadyRotated {
+						// Another instance rotated the keys between our expiry
+						// check and the update; expected when running multiple
+						// replicas against a shared database.
+						l.logger.Info("key rotation not needed", "err", err)
+					} else {
+						l.logger.Error("failed to rotate keys", "err", err)
+					}
 				}
 			}
 		}

storage/ent/client/keys.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/dexidp/dex/storage"
 	"github.com/dexidp/dex/storage/ent/db"
+	"github.com/dexidp/dex/storage/sqlretry"
 )
 
 func getKeys(ctx context.Context, client *db.KeysClient) (storage.Keys, error) {
@@ -23,7 +24,22 @@ func (d *Database) GetKeys(ctx context.Context) (storage.Keys, error) {
 }
 
 // UpdateKeys rotates keys using updater function.
+//
+// The update runs in a SERIALIZABLE transaction; when multiple dex instances
+// rotate their signing keys concurrently, the database aborts all but one
+// transaction with a serialization failure (SQLSTATE 40001/40P01). Retry the
+// whole transaction on those transient errors: every attempt re-reads the
+// current keys, and the updater's NextRotation check makes the losing instance
+// converge to errAlreadyRotated instead of failing.
 func (d *Database) UpdateKeys(ctx context.Context, updater func(old storage.Keys) (storage.Keys, error)) error {
+	return sqlretry.Do(
+		func() error { return d.updateKeysOnce(ctx, updater) },
+		sqlretry.IsSerializationFailure,
+		nil,
+	)
+}
+
+func (d *Database) updateKeysOnce(ctx context.Context, updater func(old storage.Keys) (storage.Keys, error)) error {
 	firstUpdate := false
 
 	tx, err := d.BeginTx(ctx)

storage/sql/crud.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/dexidp/dex/storage"
+	"github.com/dexidp/dex/storage/sqlretry"
 )
 
 // TODO(ericchiang): The update, insert, and select methods queries are all
@@ -418,15 +419,31 @@ func scanRefresh(s scanner) (r storage.RefreshToken, err error) {
 	return r, nil
 }
 
+// UpdateKeys rotates keys using updater function.
+//
+// The update runs in a SERIALIZABLE transaction (see flavor.executeTx); when
+// multiple dex instances rotate their signing keys concurrently, the database
+// aborts all but one transaction with a serialization failure (SQLSTATE
+// 40001/40P01). Retry the whole transaction on those transient errors: every
+// attempt re-reads the current keys, and the updater's NextRotation check
+// makes the losing instance converge to errAlreadyRotated instead of failing.
 func (c *conn) UpdateKeys(ctx context.Context, updater func(old storage.Keys) (storage.Keys, error)) error {
+	return sqlretry.Do(
+		func() error { return c.updateKeysOnce(ctx, updater) },
+		sqlretry.IsSerializationFailure,
+		nil,
+	)
+}
+
+func (c *conn) updateKeysOnce(ctx context.Context, updater func(old storage.Keys) (storage.Keys, error)) error {
 	return c.ExecTx(func(tx *trans) error {
 		firstUpdate := false
 		// TODO(ericchiang): errors may cause a transaction be rolled back by the SQL
 		// server. Test this, and consider adding a COUNT() command beforehand.
 		old, err := getKeys(ctx, tx)
 		if err != nil {
 			if err != storage.ErrNotFound {
-				return fmt.Errorf("get keys: %v", err)
+				return fmt.Errorf("get keys: %w", err)
 			}
 			firstUpdate = true
 			old = storage.Keys{}
@@ -448,7 +465,7 @@ func (c *conn) UpdateKeys(ctx context.Context, updater func(old storage.Keys) (s
 				encoder(nk.SigningKeyPub), nk.NextRotation,
 			)
 			if err != nil {
-				return fmt.Errorf("insert: %v", err)
+				return fmt.Errorf("insert: %w", err)
 			}
 		} else {
 			_, err = tx.Exec(`
@@ -464,7 +481,7 @@ func (c *conn) UpdateKeys(ctx context.Context, updater func(old storage.Keys) (s
 				encoder(nk.SigningKeyPub), nk.NextRotation, keysRowID,
 			)
 			if err != nil {
-				return fmt.Errorf("update: %v", err)
+				return fmt.Errorf("update: %w", err)
 			}
 		}
 		return nil
@@ -489,7 +506,7 @@ func getKeys(ctx context.Context, q querier) (keys storage.Keys, err error) {
 		if err == sql.ErrNoRows {
 			return keys, storage.ErrNotFound
 		}
-		return keys, fmt.Errorf("query keys: %v", err)
+		return keys, fmt.Errorf("query keys: %w", err)
 	}
 	return keys, nil
 }

storage/sqlretry/sqlretry.go

@@ -0,0 +1,85 @@
+// Package sqlretry provides a shared retry policy for transient SQL transaction
+// failures (serialization failures / deadlocks under SERIALIZABLE isolation).
+//
+// It is used by both SQL-backed storages — storage/sql (raw database/sql) and
+// storage/ent (the ent ORM) — which run signing-key rotation in a SERIALIZABLE
+// transaction and must be prepared to retry transactions the database aborts
+// under concurrency.
+package sqlretry
+
+import (
+	"errors"
+	"math/rand"
+	"time"
+
+	"github.com/go-sql-driver/mysql"
+	"github.com/lib/pq"
+)
+
+// MaxRetries is the maximum number of retries; the initial attempt is not
+// counted. Postgres requires applications to be prepared to retry transactions
+// aborted with SQLSTATE 40001; see
+// https://www.postgresql.org/docs/current/transaction-iso.html.
+//
+// 8 retries comfortably absorbs realistic signing-key rotation contention;
+// combined with jittered backoff — which de-synchronizes the retrying
+// transactions so effective concurrency at any instant stays low — it also
+// survives pathological high-contention windows, while bounding worst-case
+// latency.
+const MaxRetries = 8
+
+// backoffMs is the per-attempt base backoff in milliseconds. The last element is
+// reused for any attempt beyond its length; a random jitter of up to the same
+// magnitude is added on top to de-synchronize retrying transactions (avoid a
+// thundering herd).
+var backoffMs = []int{5, 10, 25, 50, 100}
+
+const (
+	pgErrSerializationFailure = "40001" // serialization_failure
+	pgErrDeadlockDetected     = "40P01" // deadlock_detected
+	mysqlErrLockDeadlock      = 1213    // ER_LOCK_DEADLOCK
+	mysqlErrLockWaitTimeout   = 1205    // ER_LOCK_WAIT_TIMEOUT
+)
+
+// IsSerializationFailure reports whether err is a transient transaction failure
+// (serialization failure or deadlock) that is safe to retry by re-running the
+// whole transaction. It understands both the lib/pq and go-sql-driver/mysql
+// error types, unwrapping the error chain with errors.As.
+func IsSerializationFailure(err error) bool {
+	var pqErr *pq.Error
+	if errors.As(err, &pqErr) {
+		return pqErr.Code == pgErrSerializationFailure || pqErr.Code == pgErrDeadlockDetected
+	}
+
+	var myErr *mysql.MySQLError
+	if errors.As(err, &myErr) {
+		return myErr.Number == mysqlErrLockDeadlock || myErr.Number == mysqlErrLockWaitTimeout
+	}
+
+	return false
+}
+
+// Do runs fn, retrying the whole operation on transient serialization/deadlock
+// failures with bounded, jittered backoff. Retrying is safe only when fn opens a
+// fresh transaction and re-reads current state on each attempt.
+//
+// isRetryable classifies an error as transient; if nil, no retries are performed
+// (used by backends, such as SQLite, that don't surface serialization failures).
+// onRetry, if non-nil, is called before each backoff sleep with the upcoming
+// (1-based) attempt number and the error that triggered the retry — e.g. for
+// logging.
+func Do(fn func() error, isRetryable func(error) bool, onRetry func(attempt int, err error)) error {
+	for attempt := 0; ; attempt++ {
+		err := fn()
+		if err == nil || isRetryable == nil || !isRetryable(err) || attempt >= MaxRetries {
+			return err
+		}
+
+		if onRetry != nil {
+			onRetry(attempt+1, err)
+		}
+
+		backoff := backoffMs[min(attempt, len(backoffMs)-1)]
+		time.Sleep(time.Duration(backoff+rand.Intn(backoff+1)) * time.Millisecond)
+	}
+}

storage/sqlretry/sqlretry_test.go

@@ -0,0 +1,143 @@
+package sqlretry_test
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/go-sql-driver/mysql"
+	"github.com/lib/pq"
+
+	"github.com/dexidp/dex/storage/sqlretry"
+)
+
+func TestIsSerializationFailure(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"pq serialization failure", &pq.Error{Code: "40001"}, true},
+		{"pq deadlock detected", &pq.Error{Code: "40P01"}, true},
+		{"pq unique violation", &pq.Error{Code: "23505"}, false},
+		{"mysql deadlock", &mysql.MySQLError{Number: 1213}, true},
+		{"mysql lock wait timeout", &mysql.MySQLError{Number: 1205}, true},
+		{"mysql duplicate entry", &mysql.MySQLError{Number: 1062}, false},
+		{"wrapped pq serialization failure", fmt.Errorf("update: %w", &pq.Error{Code: "40001"}), true},
+		{"plain error", errors.New("connection refused"), false},
+		{"nil error", nil, false},
+	}
+
+	for _, tt := range tests {
+		if got := sqlretry.IsSerializationFailure(tt.err); got != tt.want {
+			t.Errorf("%s: IsSerializationFailure() = %v, want %v", tt.name, got, tt.want)
+		}
+	}
+}
+
+func TestDoRetriesUntilSuccess(t *testing.T) {
+	attempts := 0
+	err := sqlretry.Do(
+		func() error {
+			attempts++
+			if attempts < 3 {
+				return &pq.Error{Code: "40001"}
+			}
+			return nil
+		},
+		sqlretry.IsSerializationFailure,
+		nil,
+	)
+	if err != nil {
+		t.Fatalf("expected success after retries, got: %v", err)
+	}
+	if attempts != 3 {
+		t.Fatalf("expected 3 attempts (2 retries), got %d", attempts)
+	}
+}
+
+func TestDoGivesUpAfterMaxRetries(t *testing.T) {
+	attempts := 0
+	sentinel := &pq.Error{Code: "40P01"}
+	err := sqlretry.Do(
+		func() error {
+			attempts++
+			return sentinel
+		},
+		sqlretry.IsSerializationFailure,
+		nil,
+	)
+	var pqErr *pq.Error
+	if !errors.As(err, &pqErr) {
+		t.Fatalf("expected the serialization failure to be returned, got: %v", err)
+	}
+	if want := sqlretry.MaxRetries + 1; attempts != want {
+		t.Fatalf("expected %d attempts (1 + %d retries), got %d", want, sqlretry.MaxRetries, attempts)
+	}
+}
+
+func TestDoDoesNotRetryNonRetryableError(t *testing.T) {
+	attempts := 0
+	sentinel := errors.New("permanent failure")
+	err := sqlretry.Do(
+		func() error {
+			attempts++
+			return sentinel
+		},
+		sqlretry.IsSerializationFailure,
+		nil,
+	)
+	if !errors.Is(err, sentinel) {
+		t.Fatalf("expected sentinel error to be returned, got: %v", err)
+	}
+	if attempts != 1 {
+		t.Fatalf("expected exactly 1 attempt for non-retryable error, got %d", attempts)
+	}
+}
+
+func TestDoNilRetryCheckDoesNotRetry(t *testing.T) {
+	attempts := 0
+	sentinel := &pq.Error{Code: "40001"}
+	err := sqlretry.Do(
+		func() error {
+			attempts++
+			return sentinel
+		},
+		nil,
+		nil,
+	)
+	var pqErr *pq.Error
+	if !errors.As(err, &pqErr) {
+		t.Fatalf("expected the serialization failure to be returned, got: %v", err)
+	}
+	if attempts != 1 {
+		t.Fatalf("expected exactly 1 attempt with nil retry check, got %d", attempts)
+	}
+}
+
+func TestDoCallsOnRetry(t *testing.T) {
+	attempts := 0
+	var retryErrors []error
+	err := sqlretry.Do(
+		func() error {
+			attempts++
+			if attempts < 2 {
+				return &pq.Error{Code: "40001"}
+			}
+			return nil
+		},
+		sqlretry.IsSerializationFailure,
+		func(attempt int, err error) {
+			if attempt != 1 {
+				t.Errorf("expected onRetry attempt 1, got %d", attempt)
+			}
+			retryErrors = append(retryErrors, err)
+		},
+	)
+	if err != nil {
+		t.Fatalf("expected success after retry, got: %v", err)
+	}
+	if len(retryErrors) != 1 {
+		t.Fatalf("expected 1 onRetry call, got %d", len(retryErrors))
+	}
+}