Re-authenticate if the session is closed by a concurrent request

kopf/_cogs/clients/api.py

@@ -84,6 +84,16 @@ async def request(
             )
             await errors.check_response(response)  # but do not parse it!
 
+        # aiohttp raises a generic error if the session/transport is closed, so we try to guess.
+        except RuntimeError as e:
+            if context.session.closed:
+                # NB: this will reset the retry counter and do the full cycle with the new creds.
+                # TODO: find a way to gracefully replace the active session in the existing context,
+                #       so that all ongoing requests would switch to new session & credentials.
+                logger.error(f"Request attempt {idx} failed; will try re-authenticating: {what}")
+                raise errors.APISessionClosed("Session is closed.") from e
+            raise
+
         except (aiohttp.ClientConnectionError, errors.APIServerError, asyncio.TimeoutError) as e:
             if backoff is None:  # i.e. the last or the only attempt.
                 logger.error(f"Request attempt {idx} failed; escalating: {what} -> {e!r}")

kopf/_cogs/clients/auth.py

@@ -43,7 +43,7 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any:
         async for key, info, context in vault.extended(APIContext, 'contexts'):
             try:
                 return await fn(*args, **kwargs, context=context)
-            except errors.APIUnauthorizedError as e:
+            except (errors.APIUnauthorizedError, errors.APISessionClosed) as e:
                 await vault.invalidate(key, exc=e)
 
         # Normally, either `vault.extended()` or `vault.invalidate()` raise the login errors.

kopf/_cogs/clients/errors.py

@@ -113,6 +113,19 @@ class APIConflictError(APIClientError):
     pass
 
 
+class APISessionClosed(Exception):
+    """
+    A helper to escalate from inside the requests to cause re-authentication.
+
+    This happens when credentials expire while multiple concurrent requests
+    are ongoing (including their retries, mostly their back-off timeouts):
+    one random request will raise HTTP 401 and cause the re-authentication,
+    while others will retry their requests with the old session (now closed!)
+    and get a generic RuntimeError from aiohttp, thus failing their whole task.
+    """
+    pass
+
+
 async def check_response(
         response: aiohttp.ClientResponse,
 ) -> None: