T 1532290: Serialize concurrent token refreshes to prevent one-time refresh token double-spend

Sources/Conduit/Auth/OAuth2RequestPipelineMiddleware.swift

@@ -23,6 +23,8 @@ public struct OAuth2RequestPipelineMiddleware: RequestPipelineMiddleware {
     /// against lengthy session locks when terminating mid-flight. This will allow the host process to quickly pick
     /// up where the other process left off, if it needs to. Defaults to 30 seconds.
     public var tokenRefreshLockRelinquishInterval: TimeInterval = 30
+    /// Kill switch for atomic in-process refresh serialization; static because instances are per-client value copies. Defaults to true.
+    public static var refreshClaimCoordinationEnabled = true
     let clientConfiguration: OAuth2ClientConfiguration
     let authorization: OAuth2Authorization
     let tokenStorage: OAuth2TokenStore
@@ -34,6 +36,11 @@ public struct OAuth2RequestPipelineMiddleware: RequestPipelineMiddleware {
         return token
     }
 
+    /// One refresh claim per client + authorization, keyed by the store's canonical token identifier.
+    private var refreshClaimKey: String {
+        tokenStorage.tokenIdentifierFor(clientConfiguration: clientConfiguration, authorization: authorization) + ".refresh-claim"
+    }
+
     /// Creates a new OAuth2RequestPipelineMiddleware
     /// - Parameters:
     ///   - clientConfiguration: The configuration of the OAuth2 client
@@ -47,6 +54,8 @@ public struct OAuth2RequestPipelineMiddleware: RequestPipelineMiddleware {
         self.tokenStorage = tokenStorage
     }
 
+    // Over the body-length limit only while the pre-coordination path stays inline; remove with it.
+    // swiftlint:disable:next function_body_length
     public func prepareForTransport(request: URLRequest, completion: @escaping (Result<URLRequest>) -> Void) {
         let url = request.url?.absoluteString ?? "(Unknown URL)"
         let method = request.httpMethod ?? "(Unknown Method)"
@@ -58,6 +67,10 @@ public struct OAuth2RequestPipelineMiddleware: RequestPipelineMiddleware {
         else if let token = token,
             token.refreshToken != nil,
             refreshStrategyFactory != nil {
+            if OAuth2RequestPipelineMiddleware.refreshClaimCoordinationEnabled {
+                refreshTokenAndResume(request: request, completion: completion)
+                return
+            }
             if tokenStorage.isRefreshTokenLockedFor(client: clientConfiguration, authorization: authorization),
                 let tokenLockExpiration = tokenStorage.refreshTokenLockExpirationFor(client: clientConfiguration, authorization: authorization) {
                 logger.info("Token refresh is active in an alternate session; retrying once lock is relinquished")
@@ -72,7 +85,7 @@ public struct OAuth2RequestPipelineMiddleware: RequestPipelineMiddleware {
             tokenStorage.lockRefreshToken(timeout: tokenRefreshLockRelinquishInterval, client: clientConfiguration, authorization: authorization)
             logger.info("Token is expired, proceeding to refresh token")
             OAuth2TokenRefreshCoordinator.shared.beginTokenRefresh()
-            
+
             refresh(token: token) { result in
                 switch result {
                 case .error(let error):
@@ -103,6 +116,105 @@ public struct OAuth2RequestPipelineMiddleware: RequestPipelineMiddleware {
         }
     }
 
+    /// Refreshes the expired token and resumes the request; cross-process coordination via the storage lock, in-process via the claim registry.
+    private func refreshTokenAndResume(request: URLRequest, completion: @escaping (Result<URLRequest>) -> Void) {
+        if tokenStorage.isRefreshTokenLockedFor(client: clientConfiguration, authorization: authorization),
+            let tokenLockExpiration = tokenStorage.refreshTokenLockExpirationFor(client: clientConfiguration, authorization: authorization) {
+            logger.info("Token refresh is active in an alternate session; retrying once lock is relinquished")
+            // Cap the wait: re-entry re-checks all state, so a missed unbuffered wake costs one short poll, not the lock lifetime.
+            let timeout = min(0.5, max(0, tokenLockExpiration.timeIntervalSinceNow))
+            waitForRefreshThenResumeOnce(timeout: timeout, request: request, completion: completion)
+            return
+        }
+        let claimKey = refreshClaimKey
+        guard let claimGeneration = OAuth2RefreshClaimRegistry.shared.tryClaim(key: claimKey,
+                                                                               staleAfter: tokenRefreshLockRelinquishInterval) else {
+            // A refresh is already in flight in this process; a second grant would double-spend the one-time token.
+            if let storedToken: BearerToken = tokenStorage.tokenFor(client: clientConfiguration, authorization: authorization),
+                storedToken.isValid {
+                // The in-flight refresh already persisted its result (store precedes release).
+                makeRequestByApplyingAuthorizationHeader(to: request, with: storedToken, completion: completion)
+                return
+            }
+            logger.info("Token refresh already in flight in this process; waiting to reuse its result")
+            // The completion notification is unbuffered; a missed wake costs one short interval, not the full lock timeout.
+            waitForRefreshThenResumeOnce(timeout: min(0.5, tokenRefreshLockRelinquishInterval),
+                                         request: request, completion: completion)
+            return
+        }
+        // Re-read after acquiring the claim: refresh the current stored token, never a pre-claim snapshot.
+        guard let currentToken: BearerToken = tokenStorage.tokenFor(client: clientConfiguration, authorization: authorization) else {
+            logger.warn("Token disappeared while acquiring the refresh claim (e.g. logout); failing request")
+            OAuth2RefreshClaimRegistry.shared.release(key: claimKey, generation: claimGeneration)
+            completion(.error(OAuth2Error.internalFailure))
+            return
+        }
+        if currentToken.isValid {
+            logger.info("Another refresh completed while acquiring the claim; reusing its token")
+            OAuth2RefreshClaimRegistry.shared.release(key: claimKey, generation: claimGeneration)
+            makeRequestByApplyingAuthorizationHeader(to: request, with: currentToken, completion: completion)
+            return
+        }
+        performRefresh(of: currentToken, claimKey: claimKey, claimGeneration: claimGeneration,
+                       request: request, completion: completion)
+    }
+
+    /// waitForRefresh can invoke its handler from both the timeout and the completion notification when
+    /// they race; collapse to a single re-entry so the request completion can never fire twice.
+    private func waitForRefreshThenResumeOnce(timeout: TimeInterval, request: URLRequest,
+                                              completion: @escaping (Result<URLRequest>) -> Void) {
+        let resumeLock = NSLock()
+        var resumed = false
+        OAuth2TokenRefreshCoordinator.shared.waitForRefresh(timeout: timeout) {
+            resumeLock.lock()
+            let alreadyResumed = resumed
+            resumed = true
+            resumeLock.unlock()
+            if alreadyResumed {
+                return
+            }
+            self.prepareForTransport(request: request, completion: completion)
+        }
+    }
+
+    /// Issues the refresh grant and resumes the request: persist, unlock, release claim, then wake waiters — in that order.
+    private func performRefresh(of token: BearerToken, claimKey: String, claimGeneration: UInt64,
+                                request: URLRequest, completion: @escaping (Result<URLRequest>) -> Void) {
+        tokenStorage.lockRefreshToken(timeout: tokenRefreshLockRelinquishInterval, client: clientConfiguration, authorization: authorization)
+        logger.info("Token is expired, proceeding to refresh token")
+        OAuth2TokenRefreshCoordinator.shared.beginTokenRefresh()
+
+        refresh(token: token) { result in
+            switch result {
+            case .error(let error):
+                logger.warn("There was an error refreshing the token")
+                // A superseded holder must not delete the successor's token or unlock its lock.
+                if OAuth2RefreshClaimRegistry.shared.owns(key: claimKey, generation: claimGeneration) {
+                    if case OAuth2Error.clientFailure = error {
+                        self.tokenStorage.removeTokenFor(client: self.clientConfiguration, authorization: self.authorization)
+                    }
+                    tokenStorage.unlockRefreshTokenFor(client: clientConfiguration, authorization: authorization)
+                }
+                OAuth2RefreshClaimRegistry.shared.release(key: claimKey, generation: claimGeneration)
+                completion(.error(error))
+            case .value(let newToken):
+                logger.info("Successfully refreshed token")
+                logger.debug("New token issued: \(newToken)")
+                // Persist before releasing the claim or waking waiters; a superseded holder must not
+                // overwrite the successor's newer token or unlock its lock.
+                if OAuth2RefreshClaimRegistry.shared.owns(key: claimKey, generation: claimGeneration) {
+                    self.tokenStorage.store(token: newToken, for: self.clientConfiguration, with: self.authorization)
+                    tokenStorage.unlockRefreshTokenFor(client: clientConfiguration, authorization: authorization)
+                }
+                OAuth2RefreshClaimRegistry.shared.release(key: claimKey, generation: claimGeneration)
+
+                /// Resume original request using freshly obtained bearer token
+                self.makeRequestByApplyingAuthorizationHeader(to: request, with: newToken, completion: completion)
+            }
+            OAuth2TokenRefreshCoordinator.shared.endTokenRefresh()
+        }
+    }
+
     private func prepareForTransportWithClientAuthorization(request: URLRequest, completion: @escaping (Result<URLRequest>) -> Void) {
         logger.verbose("No bearer token found, but client authorization is requested.")
         if authorization.type == .bearer {

Sources/Conduit/Auth/TokenStorage/Coordination/OAuth2RefreshClaimRegistry.swift

@@ -0,0 +1,53 @@
+//
+//  OAuth2RefreshClaimRegistry.swift
+//  Conduit
+//
+//  Copyright © 2026 MINDBODY. All rights reserved.
+//
+
+import Foundation
+
+/// Process-wide atomic refresh claim per client + authorization — the in-process mutual exclusion the check-then-act persisted lock lacks.
+final class OAuth2RefreshClaimRegistry {
+
+    static let shared = OAuth2RefreshClaimRegistry()
+
+    private struct Claim {
+        let generation: UInt64
+        let expiresAt: Date
+    }
+
+    private let lock = NSLock()
+    private var claims: [String: Claim] = [:]
+    private var nextGeneration: UInt64 = 0
+
+    /// Returns an ownership generation for `release(key:generation:)`, or nil if a live claim is held.
+    /// Claims expire after `staleAfter` and become stealable, so a hung refresh can never wedge refreshes permanently.
+    func tryClaim(key: String, staleAfter: TimeInterval) -> UInt64? {
+        lock.lock()
+        defer { lock.unlock() }
+        if let existing = claims[key], Date() < existing.expiresAt {
+            return nil
+        }
+        nextGeneration += 1
+        claims[key] = Claim(generation: nextGeneration, expiresAt: Date().addingTimeInterval(staleAfter))
+        return nextGeneration
+    }
+
+    /// Returns whether `generation` still owns the claim; gates completion side effects of a possibly-superseded refresh.
+    func owns(key: String, generation: UInt64) -> Bool {
+        lock.lock()
+        defer { lock.unlock() }
+        return claims[key]?.generation == generation
+    }
+
+    /// No-ops unless `generation` owns the current claim — a superseded holder cannot clear its successor's claim.
+    func release(key: String, generation: UInt64) {
+        lock.lock()
+        defer { lock.unlock() }
+        guard claims[key]?.generation == generation else {
+            return
+        }
+        claims[key] = nil
+    }
+}

Tests/ConduitTests/Auth/ConcurrentRefreshTestSupport.swift

@@ -0,0 +1,143 @@
+//
+//  ConcurrentRefreshTestSupport.swift
+//  Conduit
+//
+//  Copyright © 2026 MINDBODY. All rights reserved.
+//
+
+import Foundation
+@testable import Conduit
+
+/// Records every grant issued and returns rotated tokens like a real token endpoint.
+final class RecordingRefreshStrategyFactory: OAuth2RefreshStrategyFactory {
+    private let recordLock = NSLock()
+    private(set) var issuedGrantRefreshTokens: [String] = []
+
+    func make(refreshToken: String, clientConfiguration: OAuth2ClientConfiguration) -> OAuth2TokenGrantStrategy {
+        recordLock.lock()
+        issuedGrantRefreshTokens.append(refreshToken)
+        let rotation = issuedGrantRefreshTokens.count
+        recordLock.unlock()
+        let rotatedToken = BearerToken(accessToken: "ACCESS_\(rotation)",
+                                       refreshToken: "ROTATED_\(rotation)",
+                                       expiration: Date().addingTimeInterval(3_600))
+        return StubbedGrantStrategy(token: rotatedToken)
+    }
+}
+
+/// Completes asynchronously with a fixed token, mimicking token-endpoint latency.
+struct StubbedGrantStrategy: OAuth2TokenGrantStrategy {
+    let token: BearerToken
+
+    func issueToken(completion: @escaping (Result<BearerToken>) -> Void) {
+        DispatchQueue.global().asyncAfter(deadline: .now() + .milliseconds(50)) {
+            completion(.value(token))
+        }
+    }
+
+    func issueToken() throws -> BearerToken {
+        return token
+    }
+}
+
+/// Fails with clientFailure only after the test signals, so the claim can go stale while the grant is in flight.
+struct GatedFailureStrategy: OAuth2TokenGrantStrategy {
+    let proceed: DispatchSemaphore
+
+    func issueToken(completion: @escaping (Result<BearerToken>) -> Void) {
+        DispatchQueue.global().async {
+            self.proceed.wait()
+            completion(.error(OAuth2Error.clientFailure(nil, nil)))
+        }
+    }
+
+    func issueToken() throws -> BearerToken {
+        throw OAuth2Error.clientFailure(nil, nil)
+    }
+}
+
+final class GatedFailureStrategyFactory: OAuth2RefreshStrategyFactory {
+    let proceed = DispatchSemaphore(value: 0)
+    private let recordLock = NSLock()
+    private var recordedRefreshTokens: [String] = []
+
+    var issuedGrantRefreshTokens: [String] {
+        recordLock.lock()
+        defer { recordLock.unlock() }
+        return recordedRefreshTokens
+    }
+
+    func make(refreshToken: String, clientConfiguration: OAuth2ClientConfiguration) -> OAuth2TokenGrantStrategy {
+        recordLock.lock()
+        recordedRefreshTokens.append(refreshToken)
+        recordLock.unlock()
+        return GatedFailureStrategy(proceed: proceed)
+    }
+}
+
+/// Succeeds with a rotated token only after the test signals, so the claim can go stale while the grant is in flight.
+struct GatedSuccessStrategy: OAuth2TokenGrantStrategy {
+    let proceed: DispatchSemaphore
+    let token: BearerToken
+
+    func issueToken(completion: @escaping (Result<BearerToken>) -> Void) {
+        DispatchQueue.global().async {
+            self.proceed.wait()
+            completion(.value(token))
+        }
+    }
+
+    func issueToken() throws -> BearerToken {
+        return token
+    }
+}
+
+final class GatedSuccessStrategyFactory: OAuth2RefreshStrategyFactory {
+    let proceed = DispatchSemaphore(value: 0)
+
+    func make(refreshToken: String, clientConfiguration: OAuth2ClientConfiguration) -> OAuth2TokenGrantStrategy {
+        GatedSuccessStrategy(proceed: proceed,
+                             token: BearerToken(accessToken: "LATE", refreshToken: "RT_LATE",
+                                                expiration: Date().addingTimeInterval(3_600)))
+    }
+}
+
+/// Returns the expired token for the first read and the fresh token afterwards — scripting the
+/// "another refresh persisted while we raced" interleavings deterministically.
+final class ScriptedReadTokenStore: OAuth2TokenStore {
+    private let underlying = OAuth2TokenMemoryStore()
+    private let readLock = NSLock()
+    private var reads = 0
+    let expiredToken = BearerToken(accessToken: "EXPIRED", refreshToken: "RT_OLD", expiration: Date())
+    let freshToken = BearerToken(accessToken: "FRESH", refreshToken: "RT_NEW",
+                                 expiration: Date().addingTimeInterval(3_600))
+
+    @discardableResult
+    func store<Token: OAuth2Token & DataConvertible>(token: Token?, for client: OAuth2ClientConfiguration,
+                                                     with authorization: OAuth2Authorization) -> Bool {
+        underlying.store(token: token, for: client, with: authorization)
+    }
+
+    func tokenFor<Token: OAuth2Token & DataConvertible>(client: OAuth2ClientConfiguration,
+                                                        authorization: OAuth2Authorization) -> Token? {
+        readLock.lock()
+        defer { readLock.unlock() }
+        reads += 1
+        return (reads == 1 ? expiredToken : freshToken) as? Token
+    }
+
+    @discardableResult
+    func lockRefreshToken(timeout: TimeInterval, client: OAuth2ClientConfiguration,
+                          authorization: OAuth2Authorization) -> Bool {
+        underlying.lockRefreshToken(timeout: timeout, client: client, authorization: authorization)
+    }
+
+    @discardableResult
+    func unlockRefreshTokenFor(client: OAuth2ClientConfiguration, authorization: OAuth2Authorization) -> Bool {
+        underlying.unlockRefreshTokenFor(client: client, authorization: authorization)
+    }
+
+    func refreshTokenLockExpirationFor(client: OAuth2ClientConfiguration, authorization: OAuth2Authorization) -> Date? {
+        underlying.refreshTokenLockExpirationFor(client: client, authorization: authorization)
+    }
+}