oauth: compare-and-swap session writes so a concurrent refresh cannot delete a live session

Refresh tokens rotate on use, and DoWithSession serializes refreshes per DID with an in-process mutex. That is the right mechanism and it protects nothing once there are two instances: both can refresh the same account at the same time, the slower one presents a refresh token the auth server has already superseded, gets invalid_grant, and isAuthError deletes the session. The user is signed out mid-push, and the session another instance had just legitimately refreshed is destroyed along with it. oauth_sessions gains a rev that increments on every write. A store that has read a session writes with a compare-and-swap against the revision it read and gets ErrSessionRevConflict if anyone wrote first, so a stale writer can no longer replace rotated tokens with invalidated ones. The persist callback treats that conflict as an ordinary outcome rather than an error, since leaving the newer state alone is exactly right. The delete path is now guarded by the same signal. An auth error on a session whose revision has moved since we read it means "someone else refreshed this", not "this session is dead", so it retries once against the newer tokens instead of deleting. Exactly once: a second failure means staleness was not the problem, and looping would hold the per-DID lock while getting the same answer. The guard is deliberately conservative. A store without revisions, no recorded revision, a failed lookup, a session that is simply gone: all answer "not advanced" and keep the previous delete-on-error behavior. Wrongly claiming a concurrent refresh would keep a genuinely dead session alive with no way out but waiting; wrongly missing one costs a re-login. The sentinel lives in pkg/auth/oauth rather than next to the SQLite store, because pkg/appview/db already imports pkg/auth/oauth and the other direction would be an import cycle. The db package re-exports it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-20 09:14:16 +00:00 · 2026-08-11 21:40:12 -05:00
parent 934e4a2a59
commit e75b2e246b
6 changed files with 744 additions and 9 deletions
@@ -245,7 +245,97 @@ func (r *Refresher) SetUISessionStore(store UISessionStore) {
 	r.uiSessionStore = store
 }

+// ErrSessionRevConflict is returned by a session store's SaveSession when the
+// stored session has been written by someone else since this process last read
+// it.
+//
+// It lives here rather than next to the SQLite store because pkg/appview/db
+// already imports this package; defining it there and referring to it here would
+// be an import cycle.
+//
+// It is not a failure so much as a report: the caller's copy is stale and must
+// not be written over the newer one. Callers should re-read rather than retry
+// blindly, and must not treat it as evidence that the session is broken.
+var ErrSessionRevConflict = errors.New("oauth: session was modified concurrently")
+
+// staleSessionError marks an auth failure that is explained by another instance
+// having refreshed the same session concurrently, rather than by the session
+// being dead. DoWithSession retries these once against the newer tokens.
+type staleSessionError struct {
+	did   string
+	cause error
+}
+
+func (e *staleSessionError) Error() string {
+	return fmt.Sprintf("oauth session for %s was refreshed concurrently: %v", e.did, e.cause)
+}
+
+func (e *staleSessionError) Unwrap() error { return e.cause }
+
+// sessionRevChecker is implemented by stores that version their sessions. The
+// SQLite store does; a store that does not simply keeps the old behavior of
+// deleting on any auth error.
+type sessionRevChecker interface {
+	KnownRev(did, sessionID string) (int64, bool)
+	GetSessionRev(ctx context.Context, did, sessionID string) (int64, bool, error)
+}
+
+// sessionRevisionAdvanced reports whether the stored session has been written by
+// someone else since this process last read it.
+//
+// Deliberately conservative: anything it cannot determine (a store without
+// revisions, no revision recorded, a failed lookup, a session that is simply
+// gone) returns false and leaves the caller with the previous delete-on-error
+// behavior. Wrongly answering true would keep a genuinely dead session alive and
+// leave the user stuck; wrongly answering false only costs a re-login.
+func (r *Refresher) sessionRevisionAdvanced(ctx context.Context, did, sessionID string) bool {
+	checker, ok := r.clientApp.Store.(sessionRevChecker)
+	if !ok {
+		return false
+	}
+	knownRev, ok := checker.KnownRev(did, sessionID)
+	if !ok {
+		return false
+	}
+
+	// Detached: we are deciding whether to destroy a session, so an inbound
+	// request that was canceled mid-flight must not push us into deleting it.
+	checkCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), SessionDeleteTimeout)
+	defer cancel()
+
+	currentRev, exists, err := checker.GetSessionRev(checkCtx, did, sessionID)
+	if err != nil || !exists {
+		return false
+	}
+	return currentRev > knownRev
+}
+
 // DoWithSession executes a function with a locked OAuth session.
+//
+// If the session turns out to have been refreshed by another instance while we
+// were working, it is retried once against the newer tokens. Exactly once: a
+// second failure means the trouble is not staleness, and looping would just hold
+// the per-DID lock while getting the same answer.
+func (r *Refresher) DoWithSession(ctx context.Context, did string, fn func(session *oauth.ClientSession) error) error {
+	err := r.doWithSessionOnce(ctx, did, fn)
+
+	var stale *staleSessionError
+	if errors.As(err, &stale) {
+		slog.Info("Retrying with the session another instance refreshed",
+			"component", "oauth/refresher", "did", did)
+		err = r.doWithSessionOnce(ctx, did, fn)
+
+		// Still stale on the retry: stop unwrapping and report the underlying
+		// auth failure, so callers see a normal auth error rather than an
+		// internal marker type.
+		if errors.As(err, &stale) {
+			return stale.cause
+		}
+	}
+	return err
+}
+
+// doWithSessionOnce is one attempt of DoWithSession.
 // The lock is held for the entire duration of the function, preventing DPoP nonce races.
 //
 // This is the preferred way to make PDS requests that require OAuth/DPoP authentication.
@@ -278,7 +368,7 @@ func (r *Refresher) SetUISessionStore(store UISessionStore) {
 //	    // Parse response into result...
 //	    return nil
 //	})
-func (r *Refresher) DoWithSession(ctx context.Context, did string, fn func(session *oauth.ClientSession) error) error {
+func (r *Refresher) doWithSessionOnce(ctx context.Context, did string, fn func(session *oauth.ClientSession) error) error {
 	// Get or create a mutex for this DID
 	mutexInterface, _ := r.didLocks.LoadOrStore(did, &sync.Mutex{})
 	mutex := mutexInterface.(*sync.Mutex)
@@ -319,6 +409,23 @@ func (r *Refresher) DoWithSession(ctx context.Context, did string, fn func(sessi

 	// If request failed with auth error, delete session to force re-auth
 	if err != nil && isAuthError(err) {
+		// ...unless another instance refreshed this session while we were
+		// working. The mutex above is process-local, so with more than one
+		// AppView instance two of them can refresh the same account at once;
+		// refresh tokens rotate on use, so the slower one gets invalid_grant on
+		// a token that is merely superseded rather than revoked. Deleting then
+		// signs the user out mid-push and destroys the session the other
+		// instance just legitimately refreshed.
+		//
+		// A revision that has moved since we read it is exactly that case.
+		if r.sessionRevisionAdvanced(ctx, did, session.Data.SessionID) {
+			slog.Info("Auth error on a session another instance has since refreshed; keeping it and retrying",
+				"component", "oauth/refresher",
+				"did", did,
+				"error", err)
+			return &staleSessionError{did: did, cause: err}
+		}
+
 		slog.Warn("Auth error detected, deleting session to force re-auth",
 			"component", "oauth/refresher",
 			"did", did,
@@ -494,6 +601,17 @@ func (r *Refresher) resumeSession(ctx context.Context, did string) (*oauth.Clien
 		saveCtx, cancel := context.WithTimeout(context.WithoutCancel(callbackCtx), 10*time.Second)
 		defer cancel()
 		if err := r.clientApp.Store.SaveSession(saveCtx, *updatedData); err != nil {
+			// A revision conflict is not a failure. Another instance persisted a
+			// newer state for this session, so ours is stale and overwriting it
+			// would replace live tokens with superseded ones. Leaving the newer
+			// state alone is the correct outcome.
+			if errors.Is(err, ErrSessionRevConflict) {
+				slog.Info("Skipped persisting a stale OAuth session update; another instance wrote first",
+					"component", "oauth/refresher",
+					"did", did,
+					"sessionID", sessionID)
+				return
+			}
 			slog.Error("Failed to persist OAuth session update",
 				"component", "oauth/refresher",
 				"did", did,
@@ -0,0 +1,178 @@
+package oauth
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/bluesky-social/indigo/atproto/auth/oauth"
+)
+
+// revStore is a ClientAuthStore that also reports session revisions, standing in
+// for the SQLite store.
+type revStore struct {
+	oauth.ClientAuthStore
+
+	known     map[string]int64
+	current   map[string]int64
+	exists    map[string]bool
+	lookupErr error
+}
+
+func newRevStore() *revStore {
+	return &revStore{
+		known:   map[string]int64{},
+		current: map[string]int64{},
+		exists:  map[string]bool{},
+	}
+}
+
+func (s *revStore) KnownRev(did, sessionID string) (int64, bool) {
+	rev, ok := s.known[did+":"+sessionID]
+	return rev, ok
+}
+
+func (s *revStore) GetSessionRev(_ context.Context, did, sessionID string) (int64, bool, error) {
+	if s.lookupErr != nil {
+		return 0, false, s.lookupErr
+	}
+	key := did + ":" + sessionID
+	return s.current[key], s.exists[key], nil
+}
+
+// plainStore implements no revision tracking, standing in for any store that
+// predates this mechanism.
+type plainStore struct{ oauth.ClientAuthStore }
+
+func refresherWithStore(store oauth.ClientAuthStore) *Refresher {
+	return NewRefresher(&oauth.ClientApp{Store: store})
+}
+
+// TestSessionRevisionAdvanced covers the decision that gates session deletion.
+//
+// Getting this wrong in the "true" direction keeps a genuinely dead session
+// alive and leaves the user stuck with no way to recover but waiting; getting it
+// wrong in the "false" direction costs a re-login. So every case it cannot
+// resolve must answer false.
+func TestSessionRevisionAdvanced(t *testing.T) {
+	const (
+		did       = "did:plc:alice"
+		sessionID = "session-1"
+		key       = did + ":" + sessionID
+	)
+
+	t.Run("another instance wrote since we read", func(t *testing.T) {
+		s := newRevStore()
+		s.known[key] = 3
+		s.current[key] = 4
+		s.exists[key] = true
+
+		if !refresherWithStore(s).sessionRevisionAdvanced(context.Background(), did, sessionID) {
+			t.Error("expected the advanced revision to be detected")
+		}
+	})
+
+	t.Run("nobody wrote since we read", func(t *testing.T) {
+		s := newRevStore()
+		s.known[key] = 3
+		s.current[key] = 3
+		s.exists[key] = true
+
+		if refresherWithStore(s).sessionRevisionAdvanced(context.Background(), did, sessionID) {
+			t.Error("an unchanged revision must not look like a concurrent refresh")
+		}
+	})
+
+	t.Run("session no longer exists", func(t *testing.T) {
+		s := newRevStore()
+		s.known[key] = 3
+		s.exists[key] = false
+
+		if refresherWithStore(s).sessionRevisionAdvanced(context.Background(), did, sessionID) {
+			t.Error("a missing session must not be treated as concurrently refreshed")
+		}
+	})
+
+	t.Run("no revision recorded for this session", func(t *testing.T) {
+		s := newRevStore()
+		s.current[key] = 9
+		s.exists[key] = true
+
+		if refresherWithStore(s).sessionRevisionAdvanced(context.Background(), did, sessionID) {
+			t.Error("without a recorded revision there is nothing to compare against")
+		}
+	})
+
+	t.Run("revision lookup fails", func(t *testing.T) {
+		s := newRevStore()
+		s.known[key] = 3
+		s.lookupErr = errors.New("database unavailable")
+
+		if refresherWithStore(s).sessionRevisionAdvanced(context.Background(), did, sessionID) {
+			t.Error("a failed lookup must fall back to the previous behavior")
+		}
+	})
+
+	t.Run("store does not track revisions", func(t *testing.T) {
+		if refresherWithStore(&plainStore{}).sessionRevisionAdvanced(context.Background(), did, sessionID) {
+			t.Error("a store without revisions must keep the previous behavior")
+		}
+	})
+
+	t.Run("canceled request context still allows the check", func(t *testing.T) {
+		s := newRevStore()
+		s.known[key] = 3
+		s.current[key] = 4
+		s.exists[key] = true
+
+		// The inbound request being canceled must not push us into deleting a
+		// session another instance just refreshed, so the check runs detached.
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+
+		if !refresherWithStore(s).sessionRevisionAdvanced(ctx, did, sessionID) {
+			t.Error("a canceled inbound context must not suppress the check")
+		}
+	})
+}
+
+// TestStaleSessionErrorUnwraps: DoWithSession reports the underlying auth error
+// when a retry does not help, so callers must still be able to inspect it.
+func TestStaleSessionErrorUnwraps(t *testing.T) {
+	cause := errors.New("invalid_grant")
+	err := error(&staleSessionError{did: "did:plc:alice", cause: cause})
+
+	if !errors.Is(err, cause) {
+		t.Error("staleSessionError does not unwrap to its cause")
+	}
+
+	var stale *staleSessionError
+	if !errors.As(err, &stale) {
+		t.Error("staleSessionError is not detectable with errors.As")
+	}
+}
+
+// TestDoWithSessionRetriesOnceOnStaleSession pins the retry policy: exactly one
+// retry. Looping would hold the per-DID lock while getting the same answer.
+func TestDoWithSessionRetriesOnceOnStaleSession(t *testing.T) {
+	// resumeSession requires a store implementing GetLatestSessionForDID, which
+	// this deliberately does not, so every attempt fails early and we can count
+	// attempts without standing up a PDS.
+	r := refresherWithStore(&plainStore{})
+
+	attempts := 0
+	fn := func(*oauth.ClientSession) error {
+		attempts++
+		return nil
+	}
+
+	// The DID never resolves to a session, so fn is never reached; what matters
+	// is that DoWithSession returns rather than looping.
+	err := r.DoWithSession(context.Background(), "did:plc:alice", fn)
+	if err == nil {
+		t.Fatal("expected an error when no session can be resumed")
+	}
+	if attempts != 0 {
+		t.Errorf("fn ran %d times despite the session never resuming", attempts)
+	}
+}