Reconcile stale configured OAuth clients

docs/arch/11-auth-server-storage.md

@@ -251,7 +251,16 @@ behavior (their stored row has no auth method and reads back as a bare
 `*fosite.DefaultClient`). This brings the Redis backend in line with the
 in-memory backend, which has always enforced the pinned method.
 
-## Configuration
+## Configured-client reconciliation
+
+Operator-declared delegate clients are reconciled as one desired set at embedded auth-server startup and on a periodic heartbeat while the server runs. Each reconcile writes every client in the desired set and marks any previously-configured row no longer in it as stale. A stale row is retained for a bounded grace period before pruning, so a replacement replica does not immediately remove a client still used by an older replica. SPIFFE associations are not part of that delegate desired set: their durable rows are inert, reserved placeholders, not usable OAuth clients. Redis rows written by the configured path carry a `configured` ownership marker. A row without that marker is legacy data — adopted into the marker scheme only if its stored shape exactly matches the client being reconciled at that ID, otherwise reconciliation fails for that one entry as a genuine collision. DCR-issued and reserved rows are never touched by reconciliation, regardless of the desired set.
+
+Reconciliation is last-write-wins, with no cross-replica coordination beyond the shared Redis rows themselves: each replica's own periodic call unconditionally writes its own desired set and marks away whatever it does not want. During a rolling update, replicas running the old and new configuration briefly disagree, and a row can flip between old and new content until every old replica has been reconciled with (or replaced by) the new configuration. The stale-row grace period gives old replicas time to continue refreshing rows they still need; once no replica refreshes a removed row and the grace period elapses, a later reconcile prunes it. This is a bounded, self-healing transient and never blocks a replica's own startup or readiness. Same-ID material rotation remains last-write-wins during the rollout.
+
+Memory storage applies the same desired-set write-and-prune logic under its
+mutex, single-process, so there is no cross-replica disagreement to consider.
+DCR-issued and legacy/unowned rows remain protected there too.
+
 
 ### CRD Configuration
 

pkg/authserver/server_impl.go

@@ -49,6 +49,13 @@ type server struct {
 	// TrustedIssuers are configured (nil otherwise). Held here so Close can shut
 	// down its per-issuer JWKS refresh worker pools; nothing else releases them.
 	trustedIssuerValidator *tokenexchange.MultiIssuerTokenValidator
+	configuredReconciler   storage.ConfiguredClientReconciler
+	// configuredClients is the desired operator-configured client set,
+	// re-asserted on every reconciliation-loop tick so a live replica keeps
+	// re-publishing (and sweeping toward) its own current configuration.
+	configuredClients []fosite.Client
+	configuredCancel  context.CancelFunc
+	configuredDone    chan struct{}
 }
 
 // DefaultUpstreamFactory creates the production upstream provider based on type.
@@ -172,15 +179,11 @@ func newServer(ctx context.Context, cfg Config, stor storage.Storage) (_ *server
 		return nil, fmt.Errorf("storage backend %T does not implement storage.DCRCredentialStore", baseStore)
 	}
 
-	stor, err := decorateStorageForSPIFFE(ctx, cfg, stor)
+	stor, configuredReconciler, desiredConfiguredClients, err := setupConfiguredClients(ctx, cfg, stor)
 	if err != nil {
 		return nil, err
 	}
 
-	if err := registerDelegateClients(ctx, stor, cfg.DelegateClients); err != nil {
-		return nil, err
-	}
-
 	slog.Debug("creating OAuth2 configuration")
 
 	// Get signing key from KeyProvider
@@ -279,14 +282,113 @@ func newServer(ctx context.Context, cfg Config, stor storage.Storage) (_ *server
 		"issuer", cfg.Issuer,
 	)
 
-	return &server{
+	srv := &server{
 		handler:                router,
 		storage:                stor,
 		dcrStore:               dcrStore,
 		upstreams:              upstreams,
 		upstreamRefresher:      refresher,
 		trustedIssuerValidator: trustedIssuerValidator,
-	}, nil
+		configuredReconciler:   configuredReconciler,
+		configuredClients:      desiredConfiguredClients,
+	}
+	if configuredReconciler != nil {
+		srv.startConfiguredClientReconciliationLoop()
+	}
+	return srv, nil
+}
+
+// startConfiguredClientReconciliationLoop starts a background worker that
+// periodically re-runs ReconcileConfiguredClients with this replica's own
+// desired set on a ticker. Unlike a lease renewal, this is a full reconcile
+// every tick: a live replica re-asserts (writes) every client it wants and
+// sweeps away anything it doesn't, so a client removed from configuration
+// converges to pruned once every replica has been reconciled with (or
+// replaced by) the new configuration -- not just at startup. The worker
+// stops, and its done channel closes, once s.configuredCancel is called
+// (from Close).
+func (s *server) startConfiguredClientReconciliationLoop() {
+	renewCtx, cancel := context.WithCancel(context.Background())
+	s.configuredCancel = cancel
+	s.configuredDone = make(chan struct{})
+	done := s.configuredDone
+	go func() {
+		ticker := time.NewTicker(10 * time.Second)
+		defer ticker.Stop()
+		defer close(done)
+		for {
+			select {
+			case <-renewCtx.Done():
+				return
+			case <-ticker.C:
+				err := s.configuredReconciler.ReconcileConfiguredClients(renewCtx, s.configuredClients)
+				if err != nil {
+					slog.Warn("configured client reconciliation failed", "error", err)
+				}
+			}
+		}
+	}()
+}
+
+// setupConfiguredClients decorates storage for SPIFFE, then reconciles the
+// desired set of operator-configured (delegate) clients against durable
+// storage if the backend supports it, falling back to one-shot delegate
+// registration otherwise. It returns the (possibly wrapped) storage, the
+// reconciler the caller must use for the periodic reconciliation loop (nil if
+// the backend doesn't support reconciliation), and the desired client set the
+// loop must keep re-asserting on every tick.
+func setupConfiguredClients(ctx context.Context, cfg Config, stor storage.Storage) (
+	storage.Storage, storage.ConfiguredClientReconciler, []fosite.Client, error,
+) {
+	desiredClients, err := configuredClients(cfg)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+	stor, err = decorateStorageForSPIFFE(ctx, cfg, stor)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	baseStore := storage.Unwrap(stor)
+	reconciler, ok := baseStore.(storage.ConfiguredClientReconciler)
+	if !ok {
+		if err := registerDelegateClients(ctx, stor, cfg.DelegateClients); err != nil {
+			return nil, nil, nil, err
+		}
+		return stor, nil, nil, nil
+	}
+
+	if err := reconciler.ReconcileConfiguredClients(ctx, desiredClients); err != nil {
+		return nil, nil, nil, fmt.Errorf("reconcile configured clients: %w", err)
+	}
+	return stor, reconciler, desiredClients, nil
+}
+
+// configuredClients builds the desired set of operator-declared clients for
+// ClientRegistry reconciliation. SPIFFE static clients are deliberately
+// excluded: they are durably claimed as inert placeholders by
+// decorateStorageForSPIFFE/preflightDurableCollisions, never as the real,
+// token-exchange-capable client this function would otherwise build via
+// registry.staticClients(). Reconciling the real client here would fight
+// that placeholder for the same row (mismatched fingerprint, since the
+// placeholder carries no grant types) and, if it ever won, would durably
+// persist a usable client where only an inert stand-in must ever exist.
+func configuredClients(cfg Config) ([]fosite.Client, error) {
+	clients := make([]fosite.Client, 0, len(cfg.DelegateClients))
+	for _, delegateClient := range cfg.DelegateClients {
+		client, err := registration.NewStaticDelegateClient(registration.Config{
+			ID:         delegateClient.ClientID,
+			Secret:     delegateClient.ClientSecret,
+			GrantTypes: []string{oauthproto.GrantTypeTokenExchange},
+			Scopes:     delegateClient.Scopes,
+			Audience:   delegateClient.Audiences,
+		})
+		if err != nil {
+			return nil, fmt.Errorf("failed to create delegate client %q: %w", delegateClient.ClientID, err)
+		}
+		clients = append(clients, client)
+	}
+	return clients, nil
 }
 
 func registerDelegateClients(ctx context.Context, stor storage.Storage, delegateClients []DelegateClient) error {
@@ -537,6 +639,10 @@ func (s *server) Close() error {
 	slog.Debug("closing OAuth authorization server")
 	s.CloseIdleConnections()
 	var errs []error
+	if s.configuredCancel != nil {
+		s.configuredCancel()
+		<-s.configuredDone
+	}
 	if s.trustedIssuerValidator != nil {
 		if err := s.trustedIssuerValidator.Close(); err != nil {
 			errs = append(errs, fmt.Errorf("failed to shut down trusted-issuer validator: %w", err))

pkg/authserver/storage/memory.go

@@ -83,6 +83,11 @@ type MemoryStorage struct {
 	// marker is needed here.
 	clients map[string]fosite.Client
 
+	// configuredClients records rows explicitly owned by operator configuration.
+	// It is separate from the client value so SPIFFE placeholder behavior is not
+	// altered by an ownership marker.
+	configuredClients map[string]struct{}
+
 	// clientOrder is the least-recently-proven-used order of clients: a
 	// registration starts at the back, and RenewClientTTL moves a DCR-issued
 	// client to the back again on proven use (a successful token
@@ -235,6 +240,7 @@ func WithMinClientAge(d time.Duration) MemoryStorageOption {
 func NewMemoryStorage(opts ...MemoryStorageOption) *MemoryStorage {
 	s := &MemoryStorage{
 		clients:               make(map[string]fosite.Client),
+		configuredClients:     make(map[string]struct{}),
 		authCodes:             make(map[string]*timedEntry[fosite.Requester]),
 		accessTokens:          make(map[string]*timedEntry[fosite.Requester]),
 		refreshTokens:         make(map[string]*timedEntry[fosite.Requester]),
@@ -505,6 +511,85 @@ func (s *MemoryStorage) UpsertDCRIssuedClient(_ context.Context, client fosite.C
 	s.clientOrder = slices.DeleteFunc(s.clientOrder, func(e clientOrderEntry) bool { return e.id == id })
 	s.clientOrder = append(s.clientOrder, clientOrderEntry{id: id, touchedAt: time.Now()})
 	s.clients[id] = client
+	delete(s.configuredClients, id)
+	return nil
+}
+
+// ReconcileConfiguredClients atomically applies the complete desired configured
+// client set and removes stale explicitly configured clients.
+func (s *MemoryStorage) ReconcileConfiguredClients(_ context.Context, clients []fosite.Client) error {
+	desired := make(map[string]fosite.Client, len(clients))
+	for _, client := range clients {
+		if client == nil {
+			return fmt.Errorf("configured client is required")
+		}
+		if registration.DCRIssued(client) {
+			return fmt.Errorf("configured client %q must not carry the DCR-issued marker", client.GetID())
+		}
+		if err := ValidateRegisterableClientID(client.GetID()); err != nil {
+			return err
+		}
+		if _, exists := desired[client.GetID()]; exists {
+			return fmt.Errorf("duplicate configured client %q", client.GetID())
+		}
+		desired[client.GetID()] = client
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for id := range desired {
+		if existing, exists := s.clients[id]; exists {
+			_, configured := s.configuredClients[id]
+			if registration.DCRIssued(existing) || !configured {
+				return fmt.Errorf("%w: client %q is already registered", ErrAlreadyExists, id)
+			}
+		}
+	}
+	workingClients := make(map[string]fosite.Client, len(s.clients))
+	for id, client := range s.clients {
+		workingClients[id] = client
+	}
+	workingConfigured := make(map[string]struct{}, len(s.configuredClients))
+	for id := range s.configuredClients {
+		workingConfigured[id] = struct{}{}
+	}
+	working := &MemoryStorage{
+		clients:           workingClients,
+		configuredClients: workingConfigured,
+		clientOrder:       slices.Clone(s.clientOrder),
+		maxClients:        s.maxClients,
+		minClientAge:      s.minClientAge,
+	}
+	if err := applyConfiguredClientsLocked(working, desired); err != nil {
+		return err
+	}
+
+	s.clients = working.clients
+	s.configuredClients = working.configuredClients
+	s.clientOrder = working.clientOrder
+	return nil
+}
+
+// applyConfiguredClientsLocked applies a desired set to isolated working state.
+// The caller owns the live-state lock; working must not be shared with callers.
+func applyConfiguredClientsLocked(working *MemoryStorage, desired map[string]fosite.Client) error {
+	for id := range working.configuredClients {
+		if _, keep := desired[id]; !keep {
+			delete(working.clients, id)
+			delete(working.configuredClients, id)
+		}
+	}
+	for id, client := range desired {
+		if _, exists := working.clients[id]; exists {
+			working.clientOrder = slices.DeleteFunc(working.clientOrder, func(e clientOrderEntry) bool { return e.id == id })
+		} else if err := working.insertClientLocked(id, client); err != nil {
+			return err
+		}
+		working.clients[id] = client
+		working.configuredClients[id] = struct{}{}
+		working.clientOrder = slices.DeleteFunc(working.clientOrder, func(e clientOrderEntry) bool { return e.id == id })
+		working.clientOrder = append(working.clientOrder, clientOrderEntry{id: id, touchedAt: time.Now()})
+	}
 	return nil
 }
 
@@ -527,7 +612,11 @@ func (s *MemoryStorage) ReconcileConfiguredClient(_ context.Context, client fosi
 
 	existing, exists := s.clients[id]
 	if !exists {
-		return s.insertClientLocked(id, client)
+		if err := s.insertClientLocked(id, client); err != nil {
+			return err
+		}
+		s.configuredClients[id] = struct{}{}
+		return nil
 	}
 	if registration.DCRIssued(existing) {
 		return fmt.Errorf("%w: client %q is DCR-issued, refusing to overwrite with a configured client",
@@ -543,6 +632,7 @@ func (s *MemoryStorage) ReconcileConfiguredClient(_ context.Context, client fosi
 	s.clientOrder = slices.DeleteFunc(s.clientOrder, func(e clientOrderEntry) bool { return e.id == id })
 	s.clientOrder = append(s.clientOrder, clientOrderEntry{id: id, touchedAt: time.Now()})
 	s.clients[id] = client
+	s.configuredClients[id] = struct{}{}
 	return nil
 }
 
@@ -556,6 +646,7 @@ func (s *MemoryStorage) insertClientLocked(id string, client fosite.Client) erro
 			victim := s.clientOrder[idx].id
 			s.clientOrder = append(s.clientOrder[:idx], s.clientOrder[idx+1:]...)
 			delete(s.clients, victim)
+			delete(s.configuredClients, victim)
 			slog.Debug("evicted oldest DCR-issued client registration at capacity",
 				"client_id", victim, "max_clients", s.maxClients)
 		} else {