diff --git a/pkg/auth/oauth/client.go b/pkg/auth/oauth/client.go index 4d2f819..a8a2e7b 100644 --- a/pkg/auth/oauth/client.go +++ b/pkg/auth/oauth/client.go @@ -9,6 +9,7 @@ import ( "fmt" "log/slog" "strings" + "sync" "time" "atcr.io/pkg/atproto" @@ -146,6 +147,7 @@ type UISessionStore interface { type Refresher struct { clientApp *oauth.ClientApp uiSessionStore UISessionStore // For invalidating UI sessions on OAuth failures + didLocks sync.Map // Per-DID mutexes to prevent concurrent DPoP nonce races } // NewRefresher creates a new session refresher @@ -162,8 +164,44 @@ func (r *Refresher) SetUISessionStore(store UISessionStore) { // GetSession gets a fresh OAuth session for a DID // Loads session from database on every request (database is source of truth) +// Uses per-DID locking to prevent concurrent requests from racing on DPoP nonce updates +// +// Why locking is critical: +// During docker push, multiple layers upload concurrently. Each layer creates a new +// ClientSession by loading from database. Without locking, this race condition occurs: +// 1. Layer A loads session with stale DPoP nonce from DB +// 2. Layer B loads session with same stale nonce (A hasn't updated DB yet) +// 3. Layer A makes request → 401 "use_dpop_nonce" → gets fresh nonce → saves to DB +// 4. Layer B makes request → 401 "use_dpop_nonce" (using stale nonce from step 2) +// 5. DPoP nonce thrashing continues, eventually causing 500 errors +// +// With per-DID locking: +// 1. Layer A acquires lock, loads session, handles nonce negotiation, saves, releases lock +// 2. Layer B acquires lock AFTER A releases, loads fresh nonce from DB, succeeds func (r *Refresher) GetSession(ctx context.Context, did string) (*oauth.ClientSession, error) { - return r.resumeSession(ctx, did) + // Get or create a mutex for this DID to prevent concurrent session loads + // This prevents DPoP nonce race conditions when multiple layers upload simultaneously + mutexInterface, _ := r.didLocks.LoadOrStore(did, &sync.Mutex{}) + mutex := mutexInterface.(*sync.Mutex) + + // Serialize session loading per DID + mutex.Lock() + defer mutex.Unlock() + + slog.Debug("Acquired session lock for DID", + "component", "oauth/refresher", + "did", did) + + session, err := r.resumeSession(ctx, did) + if err != nil { + return nil, err + } + + slog.Debug("Released session lock for DID", + "component", "oauth/refresher", + "did", did) + + return session, nil } // resumeSession loads a session from storage @@ -213,8 +251,8 @@ func (r *Refresher) resumeSession(ctx context.Context, did string) (*oauth.Clien } // Set up callback to persist token updates to SQLite - // This ensures that when indigo automatically refreshes tokens, - // the new tokens are saved to the database immediately + // This ensures that when indigo automatically refreshes tokens or updates DPoP nonces, + // the new state is saved to the database immediately session.PersistSessionCallback = func(callbackCtx context.Context, updatedData *oauth.ClientSessionData) { if err := r.clientApp.Store.SaveSession(callbackCtx, *updatedData); err != nil { slog.Error("Failed to persist OAuth session update", @@ -223,10 +261,14 @@ func (r *Refresher) resumeSession(ctx context.Context, did string) (*oauth.Clien "sessionID", sessionID, "error", err) } else { - slog.Debug("Persisted OAuth token refresh to database", + // Log session updates (token refresh, DPoP nonce updates, etc.) + // Note: updatedData contains the full session state including DPoP nonce, + // but we don't log sensitive data like tokens or nonces themselves + slog.Debug("Persisted OAuth session update to database", "component", "oauth/refresher", "did", did, - "sessionID", sessionID) + "sessionID", sessionID, + "hint", "This includes token refresh and DPoP nonce updates") } } return session, nil