diff --git a/server/signer/local.go b/server/signer/local.go index a210aaa080..e362a5f732 100644 --- a/server/signer/local.go +++ b/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) + } } } } diff --git a/storage/ent/client/keys.go b/storage/ent/client/keys.go index c4e972026f..9212404258 100644 --- a/storage/ent/client/keys.go +++ b/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) diff --git a/storage/sql/crud.go b/storage/sql/crud.go index e60f29d5d5..5cd6756604 100644 --- a/storage/sql/crud.go +++ b/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,7 +419,23 @@ 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 @@ -426,7 +443,7 @@ func (c *conn) UpdateKeys(ctx context.Context, updater func(old storage.Keys) (s 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 } diff --git a/storage/sqlretry/sqlretry.go b/storage/sqlretry/sqlretry.go new file mode 100644 index 0000000000..92fc20907f --- /dev/null +++ b/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) + } +} diff --git a/storage/sqlretry/sqlretry_test.go b/storage/sqlretry/sqlretry_test.go new file mode 100644 index 0000000000..636dda506e --- /dev/null +++ b/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)) + } +}