Repository: nolar/kopf (python)

Issue:
Watchers on Custom Resources throw RuntimeError("Session is closed") and permanently die
### Long story short

Operator starts up and runs normally as expected. After for running for some time some of the watch streams will throw a `RuntimeError("Session is closed")`. Once this happens that watch stream will never restart until the operator is restarted. This only appears to happen with custom resources (configmaps are fine).

### Kopf version

1.35.6

### Kubernetes version

v1.23.13 eks

### Python version

3.9.13

### Code

```python
import kopf
import kubernetes

@kopf.on.login()
def login_fn(**kwargs):
    return kopf.login_via_client(**kwargs)

@kopf.on.startup()
def operator_configuration(settings: kopf.OperatorSettings, **kwargs):
    settings.watching.server_timeout = 300
    settings.watching.client_timeout = 300
    settings.peering.name = "operator"
    settings.peering.mandatory = True
    settings.peering.stealth = True

@kopf.on.create("MyCustomResource")
@kopf.on.update("MyCustomResource")
def mycustomresource(**_):
    print("Im Alive!")
```


### Logs

```none
[01:55:09,761] asyncio              [DEBUG   ] Getting address info mycluster.eks.amazonaws.com:443, type=<SocketKind.SOCK_STREAM: 1>, flags=<AddressInfo.AI_ADDRCONFIG: 32> took 14.625ms: [(<AddressFamily.AF_INET: 2>, <SocketKind.SOCK_STREAM: 1>, 6, '', ('1.1.1.1', 443)), (<AddressFamily.AF_INET: 2>, <SocketKind.SOCK_STREAM: 1>, 6, '', ('2.2.2.2', 443))]
[01:55:09,842] asyncio              [DEBUG   ] <asyncio.sslproto.SSLProtocol object at 0x7f13b06945e0> starts SSL handshake
[01:55:10,002] asyncio              [DEBUG   ] <asyncio.sslproto.SSLProtocol object at 0x7f13b06945e0>: SSL handshake took 159.6 ms
[01:55:10,003] asyncio              [DEBUG   ] <asyncio.TransportSocket fd=8, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('10.10.10.10', 41046), raddr=('1.1.1.1', 443)> connected to 1.1.1.1:443: (<asyncio.sslproto._SSLProtocolTransport object at 0x7f13b072b6a0>, <aiohttp.client_proto.ResponseHandler object at 0x7f13b079ae80>)
[01:55:36,674] kopf._core.engines.a [INFO    ] Re-authentication has been initiated.
[01:55:36,674] kopf.activities.auth [DEBUG   ] Activity 'login_fn' is invoked.
[01:55:36,751] asyncio              [DEBUG   ] <_SelectorSocketTransport fd=6 read=polling write=<idle, bufsize=0>> received EOF
[01:55:36,752] asyncio              [DEBUG   ] <asyncio.sslproto.SSLProtocol object at 0x7f13b06f4820> received EOF
[01:55:36,754] kopf._cogs.clients.w [ERROR   ] Request attempt #1/9 failed; will retry: GET https://mycluster.eks.amazonaws.com/apis/kopf.dev/v1/namespaces/mynamespace/kopfpeerings?watch=true&resourceVersion=13763208&timeoutSeconds=300 -> ServerDisconnectedError('Server disconnected')
[01:55:37,484] kopf.activities.auth [DEBUG   ] Client is configured via kubeconfig file.
[01:55:37,485] kopf.activities.auth [INFO    ] Activity 'login_fn' succeeded.
[01:55:37,485] kopf._core.engines.a [INFO    ] Re-authentication has finished.
[01:55:37,489] asyncio              [DEBUG   ] Get address info mycluster.eks.amazonaws.com:443, type=<SocketKind.SOCK_STREAM: 1>, flags=<AddressInfo.AI_ADDRCONFIG: 32>
[01:55:37,490] asyncio              [DEBUG   ] Getting address info mycluster.eks.amazonaws.com:443, type=<SocketKind.SOCK_STREAM: 1>, flags=<AddressInfo.AI_ADDRCONFIG: 32> took 0.790ms: [(<AddressFamily.AF_INET: 2>, <SocketKind.SOCK_STREAM: 1>, 6, '', ('1.1.1.1', 443)), (<AddressFamily.AF_INET: 2>, <SocketKind.SOCK_STREAM: 1>, 6, '', ('2.2.2.2', 443))]
[01:55:37,571] asyncio              [DEBUG   ] <asyncio.sslproto.SSLProtocol object at 0x7f13b05c57f0> starts SSL handshake
[01:55:37,730] asyncio              [DEBUG   ] <asyncio.sslproto.SSLProtocol object at 0x7f13b05c57f0>: SSL handshake took 159.4 ms
[01:55:37,731] asyncio              [DEBUG   ] <asyncio.TransportSocket fd=6, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('10.10.10.10', 44038), raddr=('1.1.1.1', 443)> connected to 1.1.1.1:443: (<asyncio.sslproto._SSLProtocolTransport object at 0x7f13b078d460>, <aiohttp.client_proto.ResponseHandler object at 0x7f13b059a640>)
[01:55:37,756] kopf._cogs.clients.w [DEBUG   ] Request attempt #2/9: GET https://mycluster.eks.amazonaws.com/apis/kopf.dev/v1/namespaces/mynamespace/kopfpeerings?watch=true&resourceVersion=13763208&timeoutSeconds=300
[01:55:37,756] kopf._cogs.clients.w [DEBUG   ] Stopping the watch-stream for kopfpeerings.v1.kopf.dev in 'mynamespace'.
[01:55:37,759] kopf._core.reactor.o [ERROR   ] Peering observer for operator@mynamespace has failed: Session is closed
Traceback (most recent call last):
  File "/venv/lib/python3.9/site-packages/kopf/_cogs/aiokits/aiotasks.py", line 108, in guard
    await coro
  File "/venv/lib/python3.9/site-packages/kopf/_core/reactor/queueing.py", line 175, in watcher
    async for raw_event in stream:
  File "/venv/lib/python3.9/site-packages/kopf/_cogs/clients/watching.py", line 82, in infinite_watch
    async for raw_event in stream:
  File "/venv/lib/python3.9/site-packages/kopf/_cogs/clients/watching.py", line 186, in continuous_watch
    async for raw_input in stream:
  File "/venv/lib/python3.9/site-packages/kopf/_cogs/clients/watching.py", line 251, in watch_objs
    async for raw_input in api.stream(
  File "/venv/lib/python3.9/site-packages/kopf/_cogs/clients/api.py", line 200, in stream
    response = await request(
  File "/venv/lib/python3.9/site-packages/kopf/_cogs/clients/auth.py", line 45, in wrapper
    return await fn(*args, **kwargs, context=context)
  File "/venv/lib/python3.9/site-packages/kopf/_cogs/clients/api.py", line 78, in request
    response = await context.session.request(
  File "/venv/lib/python3.9/site-packages/aiohttp/client.py", line 400, in _request
    raise RuntimeError("Session is closed")
RuntimeError: Session is closed
[01:56:04,118] asyncio              [DEBUG   ] Get address info mycluster.eks.amazonaws.com:443, type=<SocketKind.SOCK_STREAM: 1>, flags=<AddressInfo.AI_ADDRCONFIG: 32>
[01:56:04,119] asyncio              [DEBUG   ] Getting address info mycluster.eks.amazonaws.com:443, type=<SocketKind.SOCK_STREAM: 1>, flags=<AddressInfo.AI_ADDRCONFIG: 32> took 0.731ms: [(<AddressFamily.AF_INET: 2>, <SocketKind.SOCK_STREAM: 1>, 6, '', ('1.1.1.1', 443)), (<AddressFamily.AF_INET: 2>, <SocketKind.SOCK_STREAM: 1>, 6, '', ('2.2.2.2', 443))]
[01:56:04,200] asyncio              [DEBUG   ] <asyncio.sslproto.SSLProtocol object at 0x7f13b05c58b0> starts SSL handshake
[01:56:04,359] asyncio              [DEBUG   ] <asyncio.sslproto.SSLProtocol object at 0x7f13b05c58b0>: SSL handshake took 158.7 ms
[01:56:04,360] asyncio              [DEBUG   ] <asyncio.TransportSocket fd=7, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('10.10.10.10', 47442), raddr=('1.1.1.1', 443)> connected to 1.1.1.1:443: (<asyncio.sslproto._SSLProtocolTransport object at 0x7f13b07622e0>, <aiohttp.client_proto.ResponseHandler object at 0x7f13b06a14c0>)
[01:56:56,485] asyncio              [DEBUG   ] Get address info mycluster.eks.amazonaws.com:443, type=<SocketKind.SOCK_STREAM: 1>, flags=<AddressInfo.AI_ADDRCONFIG: 32>
[01:56:56,489] asyncio              [DEBUG   ] Getting address info mycluster.eks.amazonaws.com:443, type=<SocketKind.SOCK_STREAM: 1>, flags=<AddressInfo.AI_ADDRCONFIG: 32> took 3.814ms: [(<AddressFamily.AF_INET: 2>, <SocketKind.SOCK_STREAM: 1>, 6, '', ('1.1.1.1', 443)), (<AddressFamily.AF_INET: 2>, <SocketKind.SOCK_STREAM: 1>, 6, '', ('2.2.2.2', 443))]
[01:56:56,570] asyncio              [DEBUG   ] <asyncio.sslproto.SSLProtocol object at 0x7f13b05c5f10> starts SSL handshake
[01:56:56,730] asyncio              [DEBUG   ] <asyncio.sslproto.SSLProtocol object at 0x7f13b05c5f10>: SSL handshake took 159.4 ms
[01:56:56,730] asyncio              [DEBUG   ] <asyncio.TransportSocket fd=7, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('10.10.10.10', 53874), raddr=('1.1.1.1', 443)> connected to 1.1.1.1:443: (<asyncio.sslproto._SSLProtocolTransport object at 0x7f13b06a1b20>, <aiohttp.client_proto.ResponseHandler object at 0x7f13b07624c0>)
```


### Additional information

This is an EKS cluster and is using aws eks get-token in the kubeconfig to authenticate.

Using aiohttp version 3.8.3
using kubernetes client version: 24.2.0
aws-cli/2.2.43 Python/3.8.8 (used by kubeconfig)

Not all operators watch streams die at the same time. This is not running in a container on the cluster but on a server outside of aws.

Pull request:
Re-authenticate if the session is closed by a concurrent request
Several concurrent requests can step on each others toes while catching HTTP 401s: one of them will catch 401 and cause the re-auth normally (by invalidating the APIContext via the Vault), while others will be left with the closed session and will not be able even to try executing their retried requests to get that 401. Instead, they will get a generic RuntimeError("Session is closed") from aiohttp.

The concurrent requests are highly likely if the API returned some other errors previously, so that the requests went into the back-off sleep for a few seconds before retrying.

_A proper fix would be to retry with a new session, but we currently have no mechanisms to replace the session in a context object (safely). Hence, escalate and try again with new credentials (loosing the retry counter as a side effect)._

---

Under the hood, this concurrency caused conflicts in the vault over the state of the current and invalid credentials:

The credentials invalidation happens by the key only, which is the login handler name. It does not feed the actual credentials object that failed with 401 and thus must be remembered as broken. As a result, the 2nd, 3rd, and following failed API streams invalidated the valid credentials, which was acquired after the 1st failure and re-authentication. Therefore, the following re-authentications that tried to add the same credentials, failed — because that credentials was known to be invalid.

---

The second contributng factor was the inconsistency between the vault's credentials set (`._current`) and its boolean readiness (`._state`), protected by different locks.

Specifically, the case was this:

* Client A waits for the vault readiness state (the presence of credentials) under `Vault._ready._condition` — and gets it, and proceeds further.
* Client B gets 401, invalidates the credentials under the lock `Vault._lock`, leaves the lock, initiates the re-auth again.
* Client A reaches the lock `Vault._lock` and acquires it.
* Client A tries to get the next avalable valid credentials — but there is nothing. It fails.
* The re-authentication catches up and populates the vault with a new credentials, but it is too late.

Most likely it was overlooked before in Python 3.10-3.11 or so — because they were relatively slow and the sequence of operations managed to be fast, the event loop control was not going out of the routines fast enough on `await`. Now everything is fasy, so the control leaves the coroutines more often.

Now, all these operations are protected under the same condition object, and there should be no unprotected inconsistencies between the current credentials and the vault's boolean readiness state.

---

**TODO:** Requires tests to simulate the situation — if the hypothesis is correct.

Presumably fixes #980 #1158 

Patch:
diff --git a/kopf/_cogs/clients/api.py b/kopf/_cogs/clients/api.py
index 37e09c4..3f3ca56 100644
--- a/kopf/_cogs/clients/api.py
+++ b/kopf/_cogs/clients/api.py
@@ -85,9 +85,22 @@ async def request(
             )
             await errors.check_response(response)  # but do not parse it!
 
-        # NOTE(vsaienko): during k8s upgrade API might throw 403 forbiden. Use retries for this exception as well.
+        # aiohttp raises a generic error if the session/transport is closed, so we try to guess.
+        # NB: "session closed" will reset the retry counter and do the full cycle with the new creds.
+        except RuntimeError as e:
+            if context.session.closed:
+                # TODO: find a way to gracefully replace the active session in the existing context,
+                #       so that all ongoing requests would switch to the new session & credentials.
+                logger.error(f"Request attempt {idx} failed; TCP closed; will re-authenticate: {what}")
+                raise errors.APISessionClosed("Session is closed.") from e
+            raise
+
+        # NOTE(vsaienko): during k8s upgrade API might throw 403 forbidden. Use retries for this exception as well.
         except (aiohttp.ClientConnectionError, errors.APIServerError, asyncio.TimeoutError, errors.APIForbiddenError) as e:
-            if backoff is None:  # i.e. the last or the only attempt.
+            if '[SSL: APPLICATION_DATA_AFTER_CLOSE_NOTIFY]' in str(e):  # for ClientOSError
+                logger.error(f"Request attempt {idx} failed; SSL closed; will re-authenticate: {what}")
+                raise errors.APISessionClosed("SSL data stream is closed.") from e
+            elif backoff is None:  # i.e. the last or the only attempt.
                 logger.error(f"Request attempt {idx} failed; escalating: {what} -> {e!r}")
                 raise
             else:
diff --git a/kopf/_cogs/clients/auth.py b/kopf/_cogs/clients/auth.py
index 5b811b2..f374987 100644
--- a/kopf/_cogs/clients/auth.py
+++ b/kopf/_cogs/clients/auth.py
@@ -53,8 +53,8 @@ def authenticated(fn: _F) -> _F:
                     # Keep track of responses which are using this context.
                     context.add_response(response)
                 return response
-            except errors.APIUnauthorizedError as e:
-                await vault.invalidate(key, exc=e)
+            except (errors.APIUnauthorizedError, errors.APISessionClosed) as e:
+                await vault.invalidate(key, info, exc=e)
 
         # Normally, either `vault.extended()` or `vault.invalidate()` raise the login errors.
         # The for-cycle can only end if the yielded credentials are not invalidated before trying
diff --git a/kopf/_cogs/clients/errors.py b/kopf/_cogs/clients/errors.py
index 176f3d7..f5192ef 100644
--- a/kopf/_cogs/clients/errors.py
+++ b/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:
diff --git a/kopf/_cogs/structs/credentials.py b/kopf/_cogs/structs/credentials.py
index c46ec92..7897c2e 100644
--- a/kopf/_cogs/structs/credentials.py
+++ b/kopf/_cogs/structs/credentials.py
@@ -117,7 +117,6 @@ class Vault(AsyncIterable[tuple[VaultKey, ConnectionInfo]]):
         super().__init__()
         self._current = {}
         self._invalid = collections.defaultdict(list)
-        self._lock = asyncio.Lock()
         self._next_expiration: Optional[datetime.datetime] = None
 
         if __src is not None:
@@ -125,7 +124,8 @@ class Vault(AsyncIterable[tuple[VaultKey, ConnectionInfo]]):
 
         # Mark a pre-populated vault to be usable instantly,
         # or trigger the initial authentication for an empty vault.
-        self._ready = aiotoggles.Toggle(not self.is_empty())
+        self._guard = asyncio.Condition()
+        self._ready: bool = not self.is_empty()
 
     def __repr__(self) -> str:
         return f'<{self.__class__.__name__}: {self._current!r}>'
@@ -157,11 +157,11 @@ class Vault(AsyncIterable[tuple[VaultKey, ConnectionInfo]]):
         purpose = purpose if purpose is not None else repr(factory)
         async for key, item in self._items():
             if item.caches is None:  # quick-check with no locking overhead.
-                async with self._lock:
+                async with self._guard:
                     if item.caches is None:  # securely synchronised check.
                         item.caches = {}
             if purpose not in item.caches:  # quick-check with no locking overhead.
-                async with self._lock:
+                async with self._guard:
                     if purpose not in item.caches:  # securely synchronised check.
                         item.caches[purpose] = factory(item.info)
             yield key, item.info, cast(_T, item.caches[purpose])
@@ -182,23 +182,26 @@ class Vault(AsyncIterable[tuple[VaultKey, ConnectionInfo]]):
         # Restart on every re-authentication (if new items are added).
         while True:
 
-            # Whether on the 1st run, or during the active re-authentication,
-            # ensure that the items are ready before yielding them.
-            await self._ready.wait_for(True)
+            async with self._guard:
 
-            # Check for expiration strictly after a possible re-authentication.
-            # This might cause another re-authentication if the credentials are expired at creation.
-            await self.expire()
+                # Whether on the 1st run, or during the active re-authentication,
+                # ensure that the items are ready before yielding them.
+                await self._guard.wait_for(lambda: self._ready)
 
-            # Select the items to yield and let it (i.e. a consumer) work.
-            async with self._lock:
+                # Check for expiration strictly after a possible re-authentication.
+                # This might cause another re-authentication if the credentials are pre-expired.
+                await self._expire()
+
+                # Select the items to yield and let it (i.e. a consumer task) work.
                 yielded_key, yielded_item = self.select()
+
+            # Yield strictly outside of locks/conditions. The vault must be free for invalidations.
             yield yielded_key, yielded_item
 
             # If the yielded item has been invalidated, assume that this item has failed.
-            # Otherwise (the item is in the list), it has succeeded -- we are done.
+            # Otherwise (the item is in the list), it has succeeded -- we are done iterating.
             # Note: checked by identity, in case a similar item is re-added as a different object.
-            async with self._lock:
+            async with self._guard:
                 if yielded_key in self._current and self._current[yielded_key] is yielded_item:
                     break
 
@@ -222,37 +225,52 @@ class Vault(AsyncIterable[tuple[VaultKey, ConnectionInfo]]):
         key, item = random.choice(prioritised[top_priority])
         return key, item
 
-    async def expire(self) -> None:
+    async def expire(self) -> None:  # unused, but declared public (deprecate)
         """
         Discard the expired credentials, and re-authenticate as needed.
 
         Unlike invalidation, the expired credentials are not remembered
         and not blocked from reappearing.
         """
+        # Quick & lockless for speed: it is done on every API call, we have no time for locks.
         now = datetime.datetime.now(datetime.timezone.utc)
+        if self._next_expiration is not None and now >= self._next_expiration:
+            async with self._guard:
+                await self._expire()
 
-        # Quick & lockless for speed: it is done on every API call, we have no time for locks.
+    async def _expire(self) -> None:
+        """
+        Discard the expired credentials, and re-authenticate as needed.
+
+        Unlike invalidation, the expired credentials are not remembered
+        and not blocked from reappearing.
+        """
+        now = datetime.datetime.now(datetime.timezone.utc)
+
+        # Avoid waiting for re-auth afterwards if there is nothing to expire or change.
+        expired = False
         if self._next_expiration is not None and now >= self._next_expiration:
-            async with self._lock:
-                for key, item in list(self._current.items()):
-                    expiration = item.info.expiration
-                    if expiration is not None:
-                        if expiration.tzinfo is None:
-                            expiration = expiration.replace(tzinfo=datetime.timezone.utc)
-                        if now >= expiration:
-                            await self._flush_caches(item)
-                            del self._current[key]
-                self._update_expiration()
-                need_reauth = not self._current  # i.e. nothing is left at all
+            for key, item in list(self._current.items()):
+                expiration = item.info.expiration
+                if expiration is not None:
+                    if expiration.tzinfo is None:
+                        expiration = expiration.replace(tzinfo=datetime.timezone.utc)
+                    if now >= expiration:
+                        await self._flush_caches(item)
+                        del self._current[key]
+                        expired = True
+            self._update_expiration()
 
-            # Initiate a re-authentication activity, and block until it is finished.
-            if need_reauth:
-                await self._ready.turn_to(False)
-                await self._ready.wait_for(True)
+        # Initiate a re-authentication activity, and block until it is finished.
+        if expired and not self._current:  # i.e. nothing is left at all
+            self._ready = False
+            self._guard.notify_all()
+            await self._guard.wait_for(lambda: self._ready)
 
     async def invalidate(
             self,
             key: VaultKey,
+            info: ConnectionInfo,
             *,
             exc: Optional[Exception] = None,
     ) -> None:
@@ -271,27 +289,28 @@ class Vault(AsyncIterable[tuple[VaultKey, ConnectionInfo]]):
         The background task continues to run and tries to re-authenticate
         on the next API calls until cancelled due to the operator exit.
         """
-
         # Exclude the failed connection items from the list of available ones.
         # But keep a short history of invalid items, so that they are not re-added.
-        async with self._lock:
-            if key in self._current:
+        # The history size is estimated by the number of parallel streams trying to re-auth at once.
+        async with self._guard:
+            # Note: not "==", but "is". If not the same, then it was invalidated by other consumers,
+            # the new current credentials is something new to use (maybe equal to the old one).
+            if key in self._current and self._current[key].info is info:
                 await self._flush_caches(self._current[key])
                 self._invalid[key] = self._invalid[key][-2:] + [self._current[key]]
                 del self._current[key]
                 self._update_expiration()
-            need_reauth = not self._current  # i.e. nothing is left at all
 
-        # Initiate a re-authentication activity, and block until it is finished.
-        if need_reauth:
-            await self._ready.turn_to(False)
-            await self._ready.wait_for(True)
-
-        # If the re-auth has failed, re-raise the original exception in the current stack.
-        # If the original exception is unknown, raise normally on the next iteration's yield.
-        # The error here is optional -- for better stack traces of the original exception `exc`.
-        # Keep in mind, this routine is called in parallel from many tasks for the same keys.
-        async with self._lock:
+            # Initiate a re-authentication activity, and block until it is finished.
+            if not self._current:  # i.e. nothing is left at all
+                self._ready = False
+                self._guard.notify_all()
+                await self._guard.wait_for(lambda: self._ready)
+
+            # If the re-auth has failed, re-raise the original exception in the current stack.
+            # If the original exception is unknown, raise normally on the next iteration's yield.
+            # The error here is optional -- for better stack traces of the original exception `exc`.
+            # Keep in mind, this routine is called in parallel from many tasks for the same keys.
             if not self._current:
                 if exc is not None:
                     raise LoginError("Ran out of valid credentials. Consider installing "
@@ -309,16 +328,17 @@ class Vault(AsyncIterable[tuple[VaultKey, ConnectionInfo]]):
         from the authentication activity handlers. Some of the credentials
         can be duplicates of the existing ones -- only one of them is used then.
         """
+        async with self._guard:
 
-        # Remember the new info items (or replace the old ones). If we already see that the item
-        # is invalid (as seen in our short per-key history), we keep it as such -- this prevents
-        # consistently invalid credentials from causing infinite re-authentication again and again.
-        async with self._lock:
+            # Remember the new credentials or replace the old ones. If we already see that the item
+            # is invalid (as seen in our short per-key history), we keep it as such -- this prevents
+            # repeatedly invalid credentials from causing infinite re-authentication again & again.
             self._update_converted(__src)
 
-        # Notify the consuming tasks (client wrappers) that new credentials are ready to be used.
-        # Those tasks can be blocked in `vault.invalidate()` if there are no credentials left.
-        await self._ready.turn_to(True)
+            # Notify the consuming tasks (API clients) that new credentials are ready to be used.
+            # Those tasks can be blocked in `vault.invalidate()` if there are no credentials left.
+            self._ready = True
+            self._guard.notify_all()
 
     def is_empty(self) -> bool:
         now = datetime.datetime.now(datetime.timezone.utc)
@@ -329,16 +349,18 @@ class Vault(AsyncIterable[tuple[VaultKey, ConnectionInfo]]):
         return all(dt is not None and now >= dt for dt in expirations)  # i.e. expired
 
     async def wait_for_readiness(self) -> None:
-        await self._ready.wait_for(True)
+        async with self._guard:
+            await self._guard.wait_for(lambda: self._ready)
 
     async def wait_for_emptiness(self) -> None:
-        await self._ready.wait_for(False)
+        async with self._guard:
+            await self._guard.wait_for(lambda: not self._ready)
 
     async def close(self) -> None:
         """
         Finalize all the cached objects when the operator is ending.
         """
-        async with self._lock:
+        async with self._guard:
             for key in self._current:
                 await self._flush_caches(self._current[key])
 
diff --git a/kopf/_core/engines/activities.py b/kopf/_core/engines/activities.py
index 731fec2..3f675d4 100644
--- a/kopf/_core/engines/activities.py
+++ b/kopf/_core/engines/activities.py
@@ -74,6 +74,9 @@ async def authenticate(
         _activity_title: str = "Authentication",
 ) -> None:
     """ Retrieve the credentials once, successfully or not, and exit. """
+    # We do not need the locks protection here. There is only one activity for vault population.
+    # Even with 2+ activities, if the vault is empty, all consumers will be blocked by waiting.
+    # The API clients wake up only on the final population with the internal lock protection.
 
     # Sleep most of the time waiting for a signal to re-auth.
     await vault.wait_for_emptiness()


Test patch:
diff --git a/tests/authentication/test_vault.py b/tests/authentication/test_vault.py
index 38da004..334a5b6 100644
--- a/tests/authentication/test_vault.py
+++ b/tests/authentication/test_vault.py
@@ -28,20 +28,20 @@ async def test_not_empty_when_populated():
 
 async def test_yielding_after_creation(mocker):
     vault = Vault()
-    mocker.patch.object(vault._ready, 'wait_for')
+    mocker.patch.object(vault._guard, 'wait_for')
 
     with pytest.raises(LoginError):
         async for _, _ in vault:
             pass
 
-    assert vault._ready.wait_for.await_args_list == [((True,),)]
+    assert vault._guard.wait_for.called
 
 
 async def test_yielding_after_population(mocker):
     key1 = VaultKey('some-key')
     info1 = ConnectionInfo(server='https://expected/')
     vault = Vault()
-    mocker.patch.object(vault._ready, 'wait_for')
+    mocker.patch.object(vault._guard, 'wait_for')
 
     await vault.populate({key1: info1})
 
@@ -60,7 +60,7 @@ async def test_yielding_items_before_expiration(mocker):
     key1 = VaultKey('some-key')
     info1 = ConnectionInfo(server='https://expected/', expiration=future)
     vault = Vault()
-    mocker.patch.object(vault._ready, 'wait_for')
+    mocker.patch.object(vault._guard, 'wait_for')
 
     results = []
     await vault.populate({key1: info1})
@@ -82,7 +82,7 @@ async def test_yielding_ignores_expired_items(mocker, delta):
     info1 = ConnectionInfo(server='https://expected/', expiration=past)
     info2 = ConnectionInfo(server='https://expected/', expiration=future)
     vault = Vault()
-    mocker.patch.object(vault._ready, 'wait_for')
+    mocker.patch.object(vault._guard, 'wait_for')
 
     results = []
     await vault.populate({key1: info1, key2: info2})
@@ -101,7 +101,7 @@ async def test_yielding_when_everything_is_expired(mocker, delta):
     key1 = VaultKey('some-key')
     info1 = ConnectionInfo(server='https://expected/', expiration=past)
     vault = Vault()
-    mocker.patch.object(vault._ready, 'wait_for')
+    mocker.patch.object(vault._guard, 'wait_for')
 
     await vault.populate({key1: info1})
     with pytest.raises(LoginError):
@@ -114,27 +114,27 @@ async def test_invalidation_reraises_if_nothing_is_left_with_exception(mocker):
     key1 = VaultKey('some-key')
     info1 = ConnectionInfo(server='https://expected/')
     vault = Vault()
-    mocker.patch.object(vault._ready, 'wait_for')
+    mocker.patch.object(vault._guard, 'wait_for')
 
     await vault.populate({key1: info1})
     with pytest.raises(Exception) as e:
-        await vault.invalidate(key1, exc=exc)
+        await vault.invalidate(key1, info1, exc=exc)
 
     assert isinstance(e.value, LoginError)
     assert e.value.__cause__ is exc
-    assert vault._ready.wait_for.await_args_list == [((True,),)]
+    assert vault._guard.wait_for.called
 
 
 async def test_invalidation_continues_if_nothing_is_left_without_exception(mocker):
     key1 = VaultKey('some-key')
     info1 = ConnectionInfo(server='https://expected/')
     vault = Vault()
-    mocker.patch.object(vault._ready, 'wait_for')
+    mocker.patch.object(vault._guard, 'wait_for')
 
     await vault.populate({key1: info1})
-    await vault.invalidate(key1)
+    await vault.invalidate(key1, info1)
 
-    assert vault._ready.wait_for.await_args_list == [((True,),)]
+    assert vault._guard.wait_for.called
 
 
 async def test_invalidation_continues_if_something_is_left():
@@ -147,7 +147,7 @@ async def test_invalidation_continues_if_something_is_left():
 
     await vault.populate({key1: info1})
     await vault.populate({key2: info2})
-    await vault.invalidate(key1, exc=exc)  # no exception!
+    await vault.invalidate(key1, info1, exc=exc)  # no exception!
 
     results = []
     async for key, info in vault:
@@ -158,14 +158,33 @@ async def test_invalidation_continues_if_something_is_left():
     assert results[0][1] is info2
 
 
+async def test_invalidation_continues_if_items_is_replaced(mocker):
+    key1 = VaultKey('some-key')
+    info1 = ConnectionInfo(server='https://newer-valid-credentials/')
+    info2 = ConnectionInfo(server='https://older-expired-credentials/')
+    vault = Vault()
+    mocker.patch.object(vault._guard, 'wait_for')
+
+    await vault.populate({key1: info1})
+    await vault.invalidate(key1, info2)
+
+    results = []
+    async for key, info in vault:
+        results.append((key, info))
+
+    assert len(results) == 1
+    assert results[0][0] == key1
+    assert results[0][1] is info1
+
+
 async def test_yielding_after_invalidation(mocker):
     key1 = VaultKey('some-key')
     info1 = ConnectionInfo(server='https://expected/')
     vault = Vault()
-    mocker.patch.object(vault._ready, 'wait_for')
+    mocker.patch.object(vault._guard, 'wait_for')
 
     await vault.populate({key1: info1})
-    await vault.invalidate(key1)
+    await vault.invalidate(key1, info1)
 
     with pytest.raises(LoginError):
         async for _, _ in vault:
@@ -177,10 +196,10 @@ async def test_duplicates_are_remembered(mocker):
     info1 = ConnectionInfo(server='https://expected/')
     info2 = ConnectionInfo(server='https://expected/')  # another instance, same fields
     vault = Vault()
-    mocker.patch.object(vault._ready, 'wait_for')
+    mocker.patch.object(vault._guard, 'wait_for')
 
     await vault.populate({key1: info1})
-    await vault.invalidate(key1)
+    await vault.invalidate(key1, info1)
     await vault.populate({key1: info2})
 
     # There should be nothing to yield, despite the second populate() call.
diff --git a/tests/conftest.py b/tests/conftest.py
index a1c7a17..b128d72 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -434,7 +434,7 @@ async def _fake_vault(mocker, hostname, aresponses):
     key = VaultKey('fixture')
     info = ConnectionInfo(server=f'https://{hostname}')
     vault = Vault({key: info})
-    mocker.patch.object(vault._ready, 'wait_for')
+    mocker.patch.object(vault._guard, 'wait_for')
     try:
         yield vault
     finally: