feat(auth): ✨ add sequential credential refresh queue and availability checks

Introduce a per-provider refresh queue and background processor to serialize token refreshes and interactive re-auth flows. Changes include:
- add _refresh_queue, _queued_credentials, _unavailable_credentials, queue tracking lock and background _process_refresh_queue to gemini/iflow/qwen auth base classes
- add _queue_refresh, _ensure_queue_processor_running and is_credential_available helpers
- change proactive refresh paths to enqueue automated refreshes (preserving exponential backoff) and allow re-auth to bypass backoff when needed
- update RotatingClient to filter out credentials that are currently marked unavailable (queued for re-auth), preferring available creds but falling back if all are unavailable

This reduces concurrent refresh attempts, prevents selecting keys undergoing re-auth, and centralizes refresh/backoff behavior without changing external APIs.

src/rotator_library/client.py

@@ -630,6 +630,18 @@ class RotatingClient:
         # multiple keys have the same usage stats.
         credentials_for_provider = list(self.all_credentials[provider])
         random.shuffle(credentials_for_provider)
+        
+        # Filter out credentials that are unavailable (queued for re-auth)
+        provider_plugin = self._get_provider_instance(provider)
+        if provider_plugin and hasattr(provider_plugin, 'is_credential_available'):
+            available_creds = [
+                cred for cred in credentials_for_provider
+                if provider_plugin.is_credential_available(cred)
+            ]
+            if available_creds:
+                credentials_for_provider = available_creds
+            # If all credentials are unavailable, keep the original list
+            # (better to try unavailable creds than fail immediately)
 
         tried_creds = set()
         last_exception = None
@@ -992,6 +1004,18 @@ class RotatingClient:
         # Create a mutable copy of the keys and shuffle it.
         credentials_for_provider = list(self.all_credentials[provider])
         random.shuffle(credentials_for_provider)
+        
+        # Filter out credentials that are unavailable (queued for re-auth)
+        provider_plugin = self._get_provider_instance(provider)
+        if provider_plugin and hasattr(provider_plugin, 'is_credential_available'):
+            available_creds = [
+                cred for cred in credentials_for_provider
+                if provider_plugin.is_credential_available(cred)
+            ]
+            if available_creds:
+                credentials_for_provider = available_creds
+            # If all credentials are unavailable, keep the original list
+            # (better to try unavailable creds than fail immediately)
 
         deadline = time.time() + self.global_timeout
         tried_creds = set()

src/rotator_library/providers/gemini_auth_base.py

@@ -37,6 +37,13 @@ class GeminiAuthBase:
         # [BACKOFF TRACKING] Track consecutive failures per credential
         self._refresh_failures: Dict[str, int] = {}  # Track consecutive failures per credential
         self._next_refresh_after: Dict[str, float] = {}  # Track backoff timers (Unix timestamp)
+        
+        # [QUEUE SYSTEM] Sequential refresh processing
+        self._refresh_queue: asyncio.Queue = asyncio.Queue()
+        self._queued_credentials: set = set()  # Track credentials already in queue
+        self._unavailable_credentials: set = set()  # Mark credentials unavailable during re-auth
+        self._queue_tracking_lock = asyncio.Lock()  # Protects queue sets
+        self._queue_processor_task: Optional[asyncio.Task] = None  # Background worker task
 
     def _load_from_env(self) -> Optional[Dict[str, Any]]:
         """
@@ -342,37 +349,11 @@ class GeminiAuthBase:
             return creds
 
     async def proactively_refresh(self, credential_path: str):
-        # [BACKOFF] Check if refresh is in backoff period (matches Go's refreshFailureBackoff)
-        now = time.time()
-        if credential_path in self._next_refresh_after:
-            backoff_until = self._next_refresh_after[credential_path]
-            if now < backoff_until:
-                remaining = int(backoff_until - now)
-                lib_logger.debug(f"Skipping refresh for '{Path(credential_path).name}' (in backoff for {remaining}s)")
-                return
-
+        """Proactively refresh a credential by queueing it for refresh."""
         creds = await self._load_credentials(credential_path)
         if self._is_token_expired(creds):
-            try:
-                await self._refresh_token(credential_path, creds)
-                # [SUCCESS] Clear failure tracking on successful refresh
-                self._refresh_failures.pop(credential_path, None)
-                self._next_refresh_after.pop(credential_path, None)
-                lib_logger.debug(f"Successfully refreshed '{Path(credential_path).name}', cleared failure tracking")
-            except Exception as e:
-                # [FAILURE] Increment failure count and set exponential backoff
-                failures = self._refresh_failures.get(credential_path, 0) + 1
-                self._refresh_failures[credential_path] = failures
-
-                # Exponential backoff: 5min → 10min → 20min → 40min → max 1 hour
-                backoff_seconds = min(300 * (2 ** (failures - 1)), 3600)
-                self._next_refresh_after[credential_path] = now + backoff_seconds
-
-                lib_logger.error(
-                    f"Refresh failed for '{Path(credential_path).name}' "
-                    f"(attempt {failures}). Next retry in {backoff_seconds}s. Error: {e}"
-                )
-                # Don't re-raise - let background refresher continue with other credentials
+            # Queue for refresh with needs_reauth=False (automated refresh)
+            await self._queue_refresh(credential_path, force=False, needs_reauth=False)
 
     async def _get_lock(self, path: str) -> asyncio.Lock:
         # [FIX RACE CONDITION] Protect lock creation with a master lock
@@ -382,6 +363,92 @@ class GeminiAuthBase:
                 self._refresh_locks[path] = asyncio.Lock()
             return self._refresh_locks[path]
 
+    def is_credential_available(self, path: str) -> bool:
+        """Check if a credential is available for rotation (not queued/refreshing)."""
+        return path not in self._unavailable_credentials
+
+    async def _ensure_queue_processor_running(self):
+        """Lazily starts the queue processor if not already running."""
+        if self._queue_processor_task is None or self._queue_processor_task.done():
+            self._queue_processor_task = asyncio.create_task(self._process_refresh_queue())
+
+    async def _queue_refresh(self, path: str, force: bool = False, needs_reauth: bool = False):
+        """Add a credential to the refresh queue if not already queued.
+        
+        Args:
+            path: Credential file path
+            force: Force refresh even if not expired
+            needs_reauth: True if full re-authentication needed (bypasses backoff)
+        """
+        # IMPORTANT: Only check backoff for simple automated refreshes
+        # Re-authentication (interactive OAuth) should BYPASS backoff since it needs user input
+        if not needs_reauth:
+            now = time.time()
+            if path in self._next_refresh_after:
+                backoff_until = self._next_refresh_after[path]
+                if now < backoff_until:
+                    # Credential is in backoff for automated refresh, do not queue
+                    remaining = int(backoff_until - now)
+                    lib_logger.debug(f"Skipping automated refresh for '{Path(path).name}' (in backoff for {remaining}s)")
+                    return
+        
+        async with self._queue_tracking_lock:
+            if path not in self._queued_credentials:
+                self._queued_credentials.add(path)
+                self._unavailable_credentials.add(path)  # Mark as unavailable
+                await self._refresh_queue.put((path, force, needs_reauth))
+                await self._ensure_queue_processor_running()
+
+    async def _process_refresh_queue(self):
+        """Background worker that processes refresh requests sequentially."""
+        while True:
+            path = None
+            try:
+                # Wait for an item with timeout to allow graceful shutdown
+                try:
+                    path, force, needs_reauth = await asyncio.wait_for(
+                        self._refresh_queue.get(), 
+                        timeout=60.0
+                    )
+                except asyncio.TimeoutError:
+                    # No items for 60s, exit to save resources
+                    self._queue_processor_task = None
+                    return
+                
+                try:
+                    # Perform the actual refresh (still using per-credential lock)
+                    async with await self._get_lock(path):
+                        # Re-check if still expired (may have changed since queueing)
+                        creds = self._credentials_cache.get(path)
+                        if creds and not self._is_token_expired(creds):
+                            # No longer expired, mark as available
+                            async with self._queue_tracking_lock:
+                                self._unavailable_credentials.discard(path)
+                            continue
+                        
+                        # Perform refresh
+                        if not creds:
+                            creds = await self._load_credentials(path)
+                        await self._refresh_token(path, creds, force=force)
+                        
+                        # SUCCESS: Mark as available again
+                        async with self._queue_tracking_lock:
+                            self._unavailable_credentials.discard(path)
+                        
+                finally:
+                    # Remove from queued set
+                    async with self._queue_tracking_lock:
+                        self._queued_credentials.discard(path)
+                    self._refresh_queue.task_done()
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                lib_logger.error(f"Error in queue processor: {e}")
+                # Even on error, mark as available (backoff will prevent immediate retry)
+                if path:
+                    async with self._queue_tracking_lock:
+                        self._unavailable_credentials.discard(path)
+
     async def initialize_token(self, creds_or_path: Union[Dict[str, Any], str]) -> Dict[str, Any]:
         path = creds_or_path if isinstance(creds_or_path, str) else None
 

src/rotator_library/providers/iflow_auth_base.py

@@ -150,6 +150,13 @@ class IFlowAuthBase:
         # [BACKOFF TRACKING] Track consecutive failures per credential
         self._refresh_failures: Dict[str, int] = {}  # Track consecutive failures per credential
         self._next_refresh_after: Dict[str, float] = {}  # Track backoff timers (Unix timestamp)
+        
+        # [QUEUE SYSTEM] Sequential refresh processing
+        self._refresh_queue: asyncio.Queue = asyncio.Queue()
+        self._queued_credentials: set = set()  # Track credentials already in queue
+        self._unavailable_credentials: set = set()  # Mark credentials unavailable during re-auth
+        self._queue_tracking_lock = asyncio.Lock()  # Protects queue sets
+        self._queue_processor_task: Optional[asyncio.Task] = None  # Background worker task
 
     def _load_from_env(self) -> Optional[Dict[str, Any]]:
         """
@@ -583,36 +590,10 @@ class IFlowAuthBase:
         if not os.path.isfile(credential_identifier):
             return  # Direct API key, no refresh needed
 
-        # [BACKOFF] Check if refresh is in backoff period
-        now = time.time()
-        if credential_identifier in self._next_refresh_after:
-            backoff_until = self._next_refresh_after[credential_identifier]
-            if now < backoff_until:
-                remaining = int(backoff_until - now)
-                lib_logger.debug(f"Skipping refresh for '{Path(credential_identifier).name}' (in backoff for {remaining}s)")
-                return
-
         creds = await self._load_credentials(credential_identifier)
         if self._is_token_expired(creds):
-            try:
-                await self._refresh_token(credential_identifier)
-                # [SUCCESS] Clear failure tracking
-                self._refresh_failures.pop(credential_identifier, None)
-                self._next_refresh_after.pop(credential_identifier, None)
-                lib_logger.debug(f"Successfully refreshed '{Path(credential_identifier).name}', cleared failure tracking")
-            except Exception as e:
-                # [FAILURE] Increment failure count and set exponential backoff
-                failures = self._refresh_failures.get(credential_identifier, 0) + 1
-                self._refresh_failures[credential_identifier] = failures
-
-                # Exponential backoff: 5min → 10min → 20min → max 1 hour
-                backoff_seconds = min(300 * (2 ** (failures - 1)), 3600)
-                self._next_refresh_after[credential_identifier] = now + backoff_seconds
-
-                lib_logger.error(
-                    f"Refresh failed for '{Path(credential_identifier).name}' "
-                    f"(attempt {failures}). Next retry in {backoff_seconds}s. Error: {e}"
-                )
+            # Queue for refresh with needs_reauth=False (automated refresh)
+            await self._queue_refresh(credential_identifier, force=False, needs_reauth=False)
 
     async def _get_lock(self, path: str) -> asyncio.Lock:
         """Gets or creates a lock for the given credential path."""
@@ -622,6 +603,92 @@ class IFlowAuthBase:
                 self._refresh_locks[path] = asyncio.Lock()
             return self._refresh_locks[path]
 
+    def is_credential_available(self, path: str) -> bool:
+        """Check if a credential is available for rotation (not queued/refreshing)."""
+        return path not in self._unavailable_credentials
+
+    async def _ensure_queue_processor_running(self):
+        """Lazily starts the queue processor if not already running."""
+        if self._queue_processor_task is None or self._queue_processor_task.done():
+            self._queue_processor_task = asyncio.create_task(self._process_refresh_queue())
+
+    async def _queue_refresh(self, path: str, force: bool = False, needs_reauth: bool = False):
+        """Add a credential to the refresh queue if not already queued.
+        
+        Args:
+            path: Credential file path
+            force: Force refresh even if not expired
+            needs_reauth: True if full re-authentication needed (bypasses backoff)
+        """
+        # IMPORTANT: Only check backoff for simple automated refreshes
+        # Re-authentication (interactive OAuth) should BYPASS backoff since it needs user input
+        if not needs_reauth:
+            now = time.time()
+            if path in self._next_refresh_after:
+                backoff_until = self._next_refresh_after[path]
+                if now < backoff_until:
+                    # Credential is in backoff for automated refresh, do not queue
+                    remaining = int(backoff_until - now)
+                    lib_logger.debug(f"Skipping automated refresh for '{Path(path).name}' (in backoff for {remaining}s)")
+                    return
+        
+        async with self._queue_tracking_lock:
+            if path not in self._queued_credentials:
+                self._queued_credentials.add(path)
+                self._unavailable_credentials.add(path)  # Mark as unavailable
+                await self._refresh_queue.put((path, force, needs_reauth))
+                await self._ensure_queue_processor_running()
+
+    async def _process_refresh_queue(self):
+        """Background worker that processes refresh requests sequentially."""
+        while True:
+            path = None
+            try:
+                # Wait for an item with timeout to allow graceful shutdown
+                try:
+                    path, force, needs_reauth = await asyncio.wait_for(
+                        self._refresh_queue.get(), 
+                        timeout=60.0
+                    )
+                except asyncio.TimeoutError:
+                    # No items for 60s, exit to save resources
+                    self._queue_processor_task = None
+                    return
+                
+                try:
+                    # Perform the actual refresh (still using per-credential lock)
+                    async with await self._get_lock(path):
+                        # Re-check if still expired (may have changed since queueing)
+                        creds = self._credentials_cache.get(path)
+                        if creds and not self._is_token_expired(creds):
+                            # No longer expired, mark as available
+                            async with self._queue_tracking_lock:
+                                self._unavailable_credentials.discard(path)
+                            continue
+                        
+                        # Perform refresh
+                        if not creds:
+                            creds = await self._load_credentials(path)
+                        await self._refresh_token(path, force=force)
+                        
+                        # SUCCESS: Mark as available again
+                        async with self._queue_tracking_lock:
+                            self._unavailable_credentials.discard(path)
+                        
+                finally:
+                    # Remove from queued set
+                    async with self._queue_tracking_lock:
+                        self._queued_credentials.discard(path)
+                    self._refresh_queue.task_done()
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                lib_logger.error(f"Error in queue processor: {e}")
+                # Even on error, mark as available (backoff will prevent immediate retry)
+                if path:
+                    async with self._queue_tracking_lock:
+                        self._unavailable_credentials.discard(path)
+
     async def initialize_token(self, creds_or_path: Union[Dict[str, Any], str]) -> Dict[str, Any]:
         """
         Initiates OAuth authorization code flow if tokens are missing or invalid.

src/rotator_library/providers/qwen_auth_base.py

@@ -39,6 +39,13 @@ class QwenAuthBase:
         # [BACKOFF TRACKING] Track consecutive failures per credential
         self._refresh_failures: Dict[str, int] = {}  # Track consecutive failures per credential
         self._next_refresh_after: Dict[str, float] = {}  # Track backoff timers (Unix timestamp)
+        
+        # [QUEUE SYSTEM] Sequential refresh processing
+        self._refresh_queue: asyncio.Queue = asyncio.Queue()
+        self._queued_credentials: set = set()  # Track credentials already in queue
+        self._unavailable_credentials: set = set()  # Mark credentials unavailable during re-auth
+        self._queue_tracking_lock = asyncio.Lock()  # Protects queue sets
+        self._queue_processor_task: Optional[asyncio.Task] = None  # Background worker task
 
     def _load_from_env(self) -> Optional[Dict[str, Any]]:
         """
@@ -327,36 +334,10 @@ class QwenAuthBase:
         if not os.path.isfile(credential_identifier):
             return  # Direct API key, no refresh needed
 
-        # [BACKOFF] Check if refresh is in backoff period
-        now = time.time()
-        if credential_identifier in self._next_refresh_after:
-            backoff_until = self._next_refresh_after[credential_identifier]
-            if now < backoff_until:
-                remaining = int(backoff_until - now)
-                lib_logger.debug(f"Skipping refresh for '{Path(credential_identifier).name}' (in backoff for {remaining}s)")
-                return
-
         creds = await self._load_credentials(credential_identifier)
         if self._is_token_expired(creds):
-            try:
-                await self._refresh_token(credential_identifier)
-                # [SUCCESS] Clear failure tracking
-                self._refresh_failures.pop(credential_identifier, None)
-                self._next_refresh_after.pop(credential_identifier, None)
-                lib_logger.debug(f"Successfully refreshed '{Path(credential_identifier).name}', cleared failure tracking")
-            except Exception as e:
-                # [FAILURE] Increment failure count and set exponential backoff
-                failures = self._refresh_failures.get(credential_identifier, 0) + 1
-                self._refresh_failures[credential_identifier] = failures
-
-                # Exponential backoff: 5min → 10min → 20min → max 1 hour
-                backoff_seconds = min(300 * (2 ** (failures - 1)), 3600)
-                self._next_refresh_after[credential_identifier] = now + backoff_seconds
-
-                lib_logger.error(
-                    f"Refresh failed for '{Path(credential_identifier).name}' "
-                    f"(attempt {failures}). Next retry in {backoff_seconds}s. Error: {e}"
-                )
+            # Queue for refresh with needs_reauth=False (automated refresh)
+            await self._queue_refresh(credential_identifier, force=False, needs_reauth=False)
 
     async def _get_lock(self, path: str) -> asyncio.Lock:
         # [FIX RACE CONDITION] Protect lock creation with a master lock
@@ -365,6 +346,92 @@ class QwenAuthBase:
                 self._refresh_locks[path] = asyncio.Lock()
             return self._refresh_locks[path]
 
+    def is_credential_available(self, path: str) -> bool:
+        """Check if a credential is available for rotation (not queued/refreshing)."""
+        return path not in self._unavailable_credentials
+
+    async def _ensure_queue_processor_running(self):
+        """Lazily starts the queue processor if not already running."""
+        if self._queue_processor_task is None or self._queue_processor_task.done():
+            self._queue_processor_task = asyncio.create_task(self._process_refresh_queue())
+
+    async def _queue_refresh(self, path: str, force: bool = False, needs_reauth: bool = False):
+        """Add a credential to the refresh queue if not already queued.
+        
+        Args:
+            path: Credential file path
+            force: Force refresh even if not expired
+            needs_reauth: True if full re-authentication needed (bypasses backoff)
+        """
+        # IMPORTANT: Only check backoff for simple automated refreshes
+        # Re-authentication (interactive OAuth) should BYPASS backoff since it needs user input
+        if not needs_reauth:
+            now = time.time()
+            if path in self._next_refresh_after:
+                backoff_until = self._next_refresh_after[path]
+                if now < backoff_until:
+                    # Credential is in backoff for automated refresh, do not queue
+                    remaining = int(backoff_until - now)
+                    lib_logger.debug(f"Skipping automated refresh for '{Path(path).name}' (in backoff for {remaining}s)")
+                    return
+        
+        async with self._queue_tracking_lock:
+            if path not in self._queued_credentials:
+                self._queued_credentials.add(path)
+                self._unavailable_credentials.add(path)  # Mark as unavailable
+                await self._refresh_queue.put((path, force, needs_reauth))
+                await self._ensure_queue_processor_running()
+
+    async def _process_refresh_queue(self):
+        """Background worker that processes refresh requests sequentially."""
+        while True:
+            path = None
+            try:
+                # Wait for an item with timeout to allow graceful shutdown
+                try:
+                    path, force, needs_reauth = await asyncio.wait_for(
+                        self._refresh_queue.get(), 
+                        timeout=60.0
+                    )
+                except asyncio.TimeoutError:
+                    # No items for 60s, exit to save resources
+                    self._queue_processor_task = None
+                    return
+                
+                try:
+                    # Perform the actual refresh (still using per-credential lock)
+                    async with await self._get_lock(path):
+                        # Re-check if still expired (may have changed since queueing)
+                        creds = self._credentials_cache.get(path)
+                        if creds and not self._is_token_expired(creds):
+                            # No longer expired, mark as available
+                            async with self._queue_tracking_lock:
+                                self._unavailable_credentials.discard(path)
+                            continue
+                        
+                        # Perform refresh
+                        if not creds:
+                            creds = await self._load_credentials(path)
+                        await self._refresh_token(path, force=force)
+                        
+                        # SUCCESS: Mark as available again
+                        async with self._queue_tracking_lock:
+                            self._unavailable_credentials.discard(path)
+                        
+                finally:
+                    # Remove from queued set
+                    async with self._queue_tracking_lock:
+                        self._queued_credentials.discard(path)
+                    self._refresh_queue.task_done()
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                lib_logger.error(f"Error in queue processor: {e}")
+                # Even on error, mark as available (backoff will prevent immediate retry)
+                if path:
+                    async with self._queue_tracking_lock:
+                        self._unavailable_credentials.discard(path)
+
     async def initialize_token(self, creds_or_path: Union[Dict[str, Any], str]) -> Dict[str, Any]:
         """Initiates device flow if tokens are missing or invalid."""
         path = creds_or_path if isinstance(creds_or_path, str) else None