git.iko.soy/crush
on main help sign in
fix(auth): stop two Crush instances from invalidating each other's login

When two Crush instances shared one OAuth login (for a provider that rotates refresh tokens, like Hyper), they could each refresh the token on their own schedule. Because each refresh retires the previous token, the two instances kept revoking the credential the other was using, producing endless "authentication failed" errors even though every refresh reported success. Token refresh is now coordinated so only one refresh happens at a time, both within a single process and across processes sharing the same login. Any instance that finds the credential was just refreshed by another adopts that result instead of refreshing again, so concurrent instances converge on one working token instead of cannibalizing each other.

commitde6792039e757b533589da4165a2ed4182ed94cb authorKieran Klukas <kieran@dunkirk.sh> dateWed 24 Jun 2026, 19:07 (4 months ago) parentb10f890
2 files changed+305−46
compare with main
unifiedsplit
internal/config/refresh_singleflight_test.go+175 internal/config/store.go+130−46
internal/config/refresh_singleflight_test.gonew file+175
@@ -0,0 +1,175 @@
1+package config
2+
3+import (
4+ "context"
5+ "fmt"
6+ "os"
7+ "path/filepath"
8+ "sync"
9+ "sync/atomic"
10+ "testing"
11+ "time"
12+
13+ "github.com/charmbracelet/crush/internal/csync"
14+ "github.com/charmbracelet/crush/internal/oauth"
15+ "github.com/stretchr/testify/require"
16+)
17+
18+// newRefreshTestStore builds a ConfigStore whose hyper provider holds an
19+// expired OAuth token, persisted both in memory and on disk at configPath.
20+// Stores that share a configPath also share the per-provider refresh lock,
21+// which lets a single test process faithfully simulate two crush instances:
22+// lock.File opens a fresh descriptor per call, so two stores block each
23+// other on the same lock file exactly as two processes would.
24+func newRefreshTestStore(t *testing.T, configPath string, exchange func(ctx context.Context, providerID, refreshToken string) (*oauth.Token, error)) *ConfigStore {
25+ t.Helper()
26+
27+ expired := &oauth.Token{
28+ AccessToken: "at0",
29+ RefreshToken: "rt0",
30+ ExpiresIn: 3600,
31+ ExpiresAt: time.Now().Add(-time.Hour).Unix(),
32+ }
33+ configContent := `{
34+ "providers": {
35+ "hyper": {
36+ "oauth": {
37+ "access_token": "at0",
38+ "refresh_token": "rt0",
39+ "expires_in": 3600,
40+ "expires_at": ` + fmt.Sprintf("%d", expired.ExpiresAt) + `
41+ }
42+ }
43+ }
44+ }`
45+ require.NoError(t, os.WriteFile(configPath, []byte(configContent), 0o600))
46+
47+ providers := csync.NewMap[string, ProviderConfig]()
48+ providers.Set("hyper", ProviderConfig{
49+ ID: "hyper",
50+ Name: "Hyper",
51+ APIKey: expired.AccessToken,
52+ OAuthToken: expired,
53+ })
54+
55+ return &ConfigStore{
56+ config: &Config{Providers: providers},
57+ globalDataPath: configPath,
58+ workingDir: filepath.Dir(configPath),
59+ exchangeToken: exchange,
60+ }
61+}
62+
63+// TestRefreshOAuthToken_InProcessSingleFlight verifies that a storm of
64+// concurrent refresh calls for the same provider collapses into a single
65+// token exchange.
66+func TestRefreshOAuthToken_InProcessSingleFlight(t *testing.T) {
67+ t.Parallel()
68+
69+ configPath := filepath.Join(t.TempDir(), "crush.json")
70+
71+ var exchanges atomic.Int64
72+ store := newRefreshTestStore(t, configPath, func(ctx context.Context, providerID, refreshToken string) (*oauth.Token, error) {
73+ exchanges.Add(1)
74+ time.Sleep(50 * time.Millisecond) // hold the flight open so peers join
75+ return &oauth.Token{
76+ AccessToken: "at1",
77+ RefreshToken: "rt1",
78+ ExpiresIn: 3600,
79+ ExpiresAt: time.Now().Add(time.Hour).Unix(),
80+ }, nil
81+ })
82+
83+ const goroutines = 20
84+ var wg sync.WaitGroup
85+ start := make(chan struct{})
86+ errs := make(chan error, goroutines)
87+ for range goroutines {
88+ wg.Go(func() {
89+ <-start
90+ errs <- store.RefreshOAuthToken(context.Background(), ScopeGlobal, "hyper")
91+ })
92+ }
93+ close(start)
94+ wg.Wait()
95+ close(errs)
96+
97+ for err := range errs {
98+ require.NoError(t, err)
99+ }
100+ require.Equal(t, int64(1), exchanges.Load(), "concurrent refreshes should collapse into one exchange")
101+
102+ pc, ok := store.config.Providers.Get("hyper")
103+ require.True(t, ok)
104+ require.Equal(t, "at1", pc.OAuthToken.AccessToken)
105+ require.Equal(t, "rt1", pc.OAuthToken.RefreshToken)
106+}
107+
108+// TestRefreshOAuthToken_CrossProcessAdopt verifies that when two instances
109+// share a credential, only one performs the token exchange and the other
110+// adopts the rotated token from disk rather than reusing the consumed
111+// refresh token. The fake exchange models a rotating provider: reusing a
112+// refresh token it has already rotated returns an error, so a second
113+// exchange would be observable as a failure.
114+func TestRefreshOAuthToken_CrossProcessAdopt(t *testing.T) {
115+ t.Parallel()
116+
117+ configPath := filepath.Join(t.TempDir(), "crush.json")
118+
119+ var (
120+ mu sync.Mutex
121+ current = "rt0" // the only refresh token the server will accept
122+ exchanges atomic.Int64
123+ reuseErrors atomic.Int64
124+ )
125+ exchange := func(ctx context.Context, providerID, refreshToken string) (*oauth.Token, error) {
126+ mu.Lock()
127+ defer mu.Unlock()
128+ if refreshToken != current {
129+ reuseErrors.Add(1)
130+ return nil, fmt.Errorf("refresh token revoked")
131+ }
132+ exchanges.Add(1)
133+ time.Sleep(50 * time.Millisecond) // hold the lock so the peer must wait
134+ current = "rt1"
135+ return &oauth.Token{
136+ AccessToken: "at1",
137+ RefreshToken: "rt1",
138+ ExpiresIn: 3600,
139+ ExpiresAt: time.Now().Add(time.Hour).Unix(),
140+ }, nil
141+ }
142+
143+ // Two stores sharing the same config file and refresh lock = two
144+ // "processes".
145+ a := newRefreshTestStore(t, configPath, exchange)
146+ b := newRefreshTestStore(t, configPath, exchange)
147+
148+ var wg sync.WaitGroup
149+ start := make(chan struct{})
150+ errs := make(chan error, 2)
151+ for _, s := range []*ConfigStore{a, b} {
152+ wg.Go(func() {
153+ <-start
154+ errs <- s.RefreshOAuthToken(context.Background(), ScopeGlobal, "hyper")
155+ })
156+ }
157+ close(start)
158+ wg.Wait()
159+ close(errs)
160+
161+ for err := range errs {
162+ require.NoError(t, err)
163+ }
164+
165+ require.Equal(t, int64(1), exchanges.Load(), "only one instance should exchange")
166+ require.Equal(t, int64(0), reuseErrors.Load(), "no instance should reuse a rotated refresh token")
167+
168+ // Both instances converge on the rotated token.
169+ for name, s := range map[string]*ConfigStore{"a": a, "b": b} {
170+ pc, ok := s.config.Providers.Get("hyper")
171+ require.True(t, ok, name)
172+ require.Equal(t, "at1", pc.OAuthToken.AccessToken, name)
173+ require.Equal(t, "rt1", pc.OAuthToken.RefreshToken, name)
174+ }
175+}
internal/config/store.go+130−46
@@ -20,6 +20,7 @@ import (
2020 "github.com/charmbracelet/crush/internal/oauth/hyper"
2121 "github.com/tidwall/gjson"
2222 "github.com/tidwall/sjson"
23+ "golang.org/x/sync/singleflight"
2324)
2425
2526// configLockDeadline bounds how long lockConfig waits for the
@@ -27,6 +28,15 @@ import (
2728// honest contention; longer suggests something is wedged.
2829const configLockDeadline = 5 * time.Second
2930
31+// refreshLockDeadline bounds how long RefreshOAuthToken waits for the
32+// per-provider cross-process refresh lock. It must exceed the token
33+// exchange HTTP timeout (30s) so that a peer mid-exchange is given time
34+// to finish and publish its result, which we then adopt instead of
35+// running our own exchange. Running our own would reuse an
36+// already-rotated refresh token and trip the provider's reuse detection,
37+// revoking the whole token family.
38+const refreshLockDeadline = 45 * time.Second
39+
3040// fileSnapshot captures metadata about a config file at a point in time.
3141type fileSnapshot struct {
3242 Path string
@@ -80,6 +90,18 @@ type ConfigStore struct {
8090
8191 mu sync.Mutex // serialises config file writes
8292 writeMu sync.Mutex // serialises in-memory config production (mutators + reload)
93+
94+ // refreshSF collapses concurrent in-process OAuth refreshes for the
95+ // same provider into a single attempt. Combined with the per-provider
96+ // cross-process refresh lock, it ensures only one token exchange runs
97+ // at a time. See RefreshOAuthToken.
98+ refreshSF singleflight.Group
99+
100+ // exchangeToken performs the provider-specific OAuth token exchange.
101+ // It is a field so tests can substitute a fake exchange without making
102+ // real network calls. Production code leaves it nil, and exchange falls
103+ // back to the real provider clients.
104+ exchangeToken func(ctx context.Context, providerID, refreshToken string) (*oauth.Token, error)
83105}
84106
85107// Config returns the pure-data config struct (read-only after load).
@@ -480,64 +502,82 @@ func (s *ConfigStore) SetProviderAPIKey(scope Scope, providerID string, apiKey a
480502
481503// RefreshOAuthToken refreshes the OAuth token for the given provider.
482504//
483−// It uses two-phase locking: the pre-check (reading the config file to
484−// see if another process already refreshed) happens under the config
485−// lock, then the HTTP exchange runs without any lock held, and finally
486−// the result is persisted via SetConfigFields (which acquires the lock
487−// internally). If the exchange fails — e.g. because another process
488−// already rotated the refresh token — the disk is re-checked under lock
489−// to recover the other process's token.
505+// Providers like Hyper rotate refresh tokens: each exchange consumes the
506+// caller's refresh token, issues a new pair, and revokes the old one. If
507+// two crush instances (or two goroutines) refresh concurrently with the
508+// same stored refresh token, the second exchange reuses an already-revoked
509+// token, trips the provider's reuse detection, and revokes the entire
510+// token family — leaving both with dead tokens even though each refresh
511+// "succeeded".
512+//
513+// To prevent that, refreshes are single-flighted at two levels:
514+//
515+// - In-process: refreshSF collapses concurrent goroutines for the same
516+// provider into one attempt.
517+// - Cross-process: a per-provider advisory lock is held across the whole
518+// read-decide-exchange-write cycle, so only one process exchanges at a
519+// time. A process that acquires the lock after a peer rotated finds the
520+// peer's fresh token on disk and adopts it instead of exchanging.
490521func (s *ConfigStore) RefreshOAuthToken(ctx context.Context, scope Scope, providerID string) error {
522+ key := fmt.Sprintf("%d\x00%s", scope, providerID)
523+ _, err, _ := s.refreshSF.Do(key, func() (any, error) {
524+ return nil, s.refreshOAuthTokenLocked(ctx, scope, providerID)
525+ })
526+ return err
527+}
528+
529+// refreshOAuthTokenLocked performs the cross-process single-flighted
530+// refresh. It is invoked through refreshSF, so at most one goroutine per
531+// provider runs it at a time within this process.
532+func (s *ConfigStore) refreshOAuthTokenLocked(ctx context.Context, scope Scope, providerID string) error {
491533 cfg := s.Config()
492534 providerConfig, exists := cfg.Providers.Get(providerID)
493535 if !exists {
494536 return fmt.Errorf("provider %s not found", providerID)
495537 }
496−
497538 if providerConfig.OAuthToken == nil {
498539 return fmt.Errorf("provider %s does not have an OAuth token", providerID)
499540 }
541+ entryToken := providerConfig.OAuthToken
500542
501− // Phase 1: Pre-check under lock — did another process already refresh?
502− release, lockErr := s.lockConfig(scope)
543+ // Acquire the per-provider cross-process refresh lock. This is a
544+ // dedicated lock file, not the config-write lock, and it does not take
545+ // s.mu — so the network exchange below cannot stall unrelated config
546+ // operations. The deadline exceeds the exchange timeout so that a peer
547+ // mid-exchange has time to publish a token we can adopt. Lock ordering:
548+ // the refresh lock is always taken before the config-write lock (via
549+ // SetConfigFields), never the reverse, so no deadlock is possible.
550+ lockCtx, cancel := context.WithTimeout(ctx, refreshLockDeadline)
551+ defer cancel()
552+ release, lockErr := lock.File(lockCtx, s.refreshLockPath(providerID))
503553 if lockErr != nil {
504− slog.Warn("Failed to lock config for pre-check, proceeding anyway", "provider", providerID, "error", lockErr)
505− } else {
506− diskToken, err := s.loadTokenFromDisk(scope, providerID)
507− release()
508− if err != nil {
509− slog.Warn("Failed to read token from config file", "provider", providerID, "error", err)
510− } else if diskToken != nil && !diskToken.IsExpired() && diskToken.AccessToken != providerConfig.OAuthToken.AccessToken {
511− slog.Info("Using token refreshed by another session", "provider", providerID)
554+ // Could not acquire the lock (peer wedged or deadline hit). Prefer a
555+ // usable token already on disk over forcing our own exchange, which
556+ // would risk reusing a rotated refresh token.
557+ if diskToken := s.adoptableDiskToken(scope, providerID, entryToken); diskToken != nil {
558+ slog.Warn("Refresh lock unavailable; adopting token from disk", "provider", providerID, "error", lockErr)
512559 return s.applyToken(providerConfig, diskToken, providerID)
513560 }
561+ return fmt.Errorf("acquire refresh lock for provider %s: %w", providerID, lockErr)
514562 }
563+ defer release()
515564
516− // Phase 2: HTTP exchange — no lock held.
517− var refreshedToken *oauth.Token
518− var refreshErr error
519− switch providerID {
520− case string(catwalk.InferenceProviderCopilot):
521− refreshedToken, refreshErr = copilot.RefreshToken(ctx, providerConfig.OAuthToken.RefreshToken)
522− case hyperp.Name:
523− refreshedToken, refreshErr = hyper.ExchangeToken(ctx, providerConfig.OAuthToken.RefreshToken)
524− default:
525− return fmt.Errorf("OAuth refresh not supported for provider %s", providerID)
565+ // Did a peer rotate the token while we waited for the lock? If disk now
566+ // holds a different, unexpired token, adopt it instead of exchanging.
567+ if diskToken := s.adoptableDiskToken(scope, providerID, entryToken); diskToken != nil {
568+ slog.Info("Adopting token refreshed by another session", "provider", providerID)
569+ return s.applyToken(providerConfig, diskToken, providerID)
526570 }
571+
572+ // Disk still holds our token (or no usable peer token exists) and we hold
573+ // the lock, so we are the sole exchanger. Perform the exchange.
574+ refreshedToken, refreshErr := s.exchange(ctx, providerID, entryToken.RefreshToken)
527575 if refreshErr != nil {
528− // Phase 3: Fallback — re-check disk under lock. The exchange may
529− // have failed because another process already rotated the refresh
530− // token.
531− if release, lockErr := s.lockConfig(scope); lockErr == nil {
532− diskToken, diskErr := s.loadTokenFromDisk(scope, providerID)
533− release()
534− if diskErr == nil &&
535− diskToken != nil &&
536− !diskToken.IsExpired() &&
537− diskToken.AccessToken != providerConfig.OAuthToken.AccessToken {
538− slog.Info("Using token refreshed by another session after exchange failure", "provider", providerID)
539− return s.applyToken(providerConfig, diskToken, providerID)
540− }
576+ // The exchange may have failed because a peer rotated the refresh
577+ // token in a window we did not cover. Re-check disk and adopt.
578+ if diskToken := s.adoptableDiskToken(scope, providerID, entryToken); diskToken != nil {
579+ slog.Info("Adopting token refreshed by another session after exchange failure", "provider", providerID)
580+ return s.applyToken(providerConfig, diskToken, providerID)
541581 }
542582 return fmt.Errorf("failed to refresh OAuth token for provider %s: %w", providerID, refreshErr)
543583 }
@@ -545,12 +585,9 @@ func (s *ConfigStore) RefreshOAuthToken(ctx context.Context, scope Scope, provid
545585 slog.Info("Successfully refreshed OAuth token", "provider", providerID)
546586 providerConfig.OAuthToken = refreshedToken
547587 providerConfig.APIKey = refreshedToken.AccessToken
548−
549− switch providerID {
550− case string(catwalk.InferenceProviderCopilot):
588+ if providerID == string(catwalk.InferenceProviderCopilot) {
551589 providerConfig.SetupGitHubCopilot()
552590 }
553−
554591 cfg.Providers.Set(providerID, providerConfig)
555592
556593 if err := s.SetConfigFields(scope, map[string]any{
@@ -559,10 +596,57 @@ func (s *ConfigStore) RefreshOAuthToken(ctx context.Context, scope Scope, provid
559596 }); err != nil {
560597 return fmt.Errorf("failed to persist refreshed token: %w", err)
561598 }
562−
563599 return nil
564600}
565601
602+// adoptableDiskToken returns the on-disk token for the provider when it is
603+// usable and differs from entryToken — i.e. when another session has
604+// already refreshed it and we should adopt that result rather than running
605+// our own exchange. It returns nil when there is nothing newer to adopt.
606+func (s *ConfigStore) adoptableDiskToken(scope Scope, providerID string, entryToken *oauth.Token) *oauth.Token {
607+ diskToken, err := s.loadTokenFromDisk(scope, providerID)
608+ if err != nil {
609+ slog.Warn("Failed to read token from config file", "provider", providerID, "error", err)
610+ return nil
611+ }
612+ if diskToken == nil || diskToken.IsExpired() {
613+ return nil
614+ }
615+ if diskToken.AccessToken == entryToken.AccessToken {
616+ // Same token we started with; nobody refreshed since.
617+ return nil
618+ }
619+ return diskToken
620+}
621+
622+// exchange performs the provider-specific OAuth token exchange. Tests may
623+// override it via the exchangeToken field; production uses the real
624+// provider clients.
625+func (s *ConfigStore) exchange(ctx context.Context, providerID, refreshToken string) (*oauth.Token, error) {
626+ if s.exchangeToken != nil {
627+ return s.exchangeToken(ctx, providerID, refreshToken)
628+ }
629+ switch providerID {
630+ case string(catwalk.InferenceProviderCopilot):
631+ return copilot.RefreshToken(ctx, refreshToken)
632+ case hyperp.Name:
633+ return hyper.ExchangeToken(ctx, refreshToken)
634+ default:
635+ return nil, fmt.Errorf("OAuth refresh not supported for provider %s", providerID)
636+ }
637+}
638+
639+// refreshLockPath returns the path to the per-provider cross-process refresh
640+// lock file. Lock files live under a dedicated locks/ subdirectory of the
641+// data dir so they do not clutter the config directory. The file is created
642+// on demand by lock.File and is never removed (flock keys on inode, not
643+// path).
644+func (s *ConfigStore) refreshLockPath(providerID string) string {
645+ dir := filepath.Join(filepath.Dir(s.globalDataPath), "locks")
646+ _ = os.MkdirAll(dir, 0o755)
647+ return filepath.Join(dir, fmt.Sprintf("%s.refresh.lock", providerID))
648+}
649+
566650// applyToken updates the in-memory provider config with the given token.
567651func (s *ConfigStore) applyToken(providerConfig ProviderConfig, token *oauth.Token, providerID string) error {
568652 providerConfig.OAuthToken = token