bitbake: hashserv: client: Fix changing stream modes

When switching from normal mode to stream mode, skip calling
self._set_mode() again because this will cause a recursion into the
_set_mode() function and causes problems.

Also cleanup some of the error checking during this process

This bug affected when a client would attempt to switch from one stream
mode to another, and meant that the server would get an invalid message
from the client. This would cause the server to disconnect the client,
and the client would then reconnect in normal mode which was the mode it
wanted anyway and thus it would carry on without any errors. This made
the bug not visible on the client side, but resulting in a lot of
backtrace JSON decoding exceptions in the server logs.

(Bitbake rev: 1826bc41ab3369ac40034c5eaf698748b769b881)

Signed-off-by: Joshua Watt <JPEWhacker@gmail.com>
Signed-off-by: Richard Purdie <richard.purdie@linuxfoundation.org>

1 files changed, 8 insertions, 4 deletions

diff --git a/bitbake/lib/hashserv/client.py b/bitbake/lib/hashserv/client.py
index d415617b20..a510f3284f 100644
--- a/bitbake/lib/hashserv/client.py
+++ b/bitbake/lib/hashserv/client.py
@@ -121,24 +121,28 @@ class AsyncClient(bb.asyncrpc.AsyncClient):
 
         return await self._send_wrapper(proc)
 
-    async def invoke(self, *args, **kwargs):
+    async def invoke(self, *args, skip_mode=False, **kwargs):
         # It's OK if connection errors cause a failure here, because the mode
         # is also reset to normal on a new connection
-        await self._set_mode(self.MODE_NORMAL)
+        if not skip_mode:
+            await self._set_mode(self.MODE_NORMAL)
         return await super().invoke(*args, **kwargs)
 
     async def _set_mode(self, new_mode):
         async def stream_to_normal():
+            # Check if already in normal mode (e.g. due to a connection reset)
+            if self.mode == self.MODE_NORMAL:
+                return "ok"
             await self.socket.send("END")
             return await self.socket.recv()
 
         async def normal_to_stream(command):
-            r = await self.invoke({command: None})
+            r = await self.invoke({command: None}, skip_mode=True)
             if r != "ok":
+                self.check_invoke_error(r)
                 raise ConnectionError(
                     f"Unable to transition to stream mode: Bad response from server {r!r}"
                 )
-
             self.logger.debug("Mode is now %s", command)
 
         if new_mode == self.mode: