Merge pull request #269 from Obihoernchen/fix/dangling-asyncio-tasks

Keep spawned asyncio tasks referenced

confluent_client/bin/nodeconsole

@@ -254,14 +254,23 @@ class InputHandler:
         currloop = asyncio.get_running_loop()
         self.fd = sys.stdin.fileno()
         self.seqtimeout = None
+        self.pendinginput = asyncio.Queue()
+        self.inputconsumer = asyncio.create_task(self.consume_input())
         currloop.add_reader(self.fd, self.handle_input)
         return self
-    
+
     def handle_input(self):
         data = os.read(self.fd, 1)
         if not data:
             return
-        asyncio.create_task(self.process_input(data))
+        self.pendinginput.put_nowait(data)
+
+    async def consume_input(self):
+        # One consumer, so a keystroke that blocks relaying cannot be overtaken
+        # by the next one, and the sequence buffer is only touched by one task
+        while True:
+            data = await self.pendinginput.get()
+            await self.process_input(data)
 
     async def timeout_sequence(self, timeout=3, flushbuffer=False):
         await asyncio.sleep(timeout)
@@ -814,15 +823,17 @@ def redraw():
             sys.stdout.write('\n')
         sys.stdout.flush()
 resized = False
+inputwatcher = None
 
 async def do_screenshot():
     global streaming
     global resized
     global numrows
     global firstnodename
+    global inputwatcher
     sess = client.Command()
     if streaming:
-        asyncio.create_task(watch_input())
+        inputwatcher = asyncio.create_task(watch_input())
     if options.tile:
         imageformat = os.environ.get('CONFLUENT_IMAGE_PROTOCOL', 'kitty')
         if imageformat not in ('kitty', 'iterm'):

confluent_server/confluent/asynchttp.py

@@ -66,6 +66,7 @@ class AsyncSession(object):
         self.wshandler = wshandler
         self.termrelations = []
         self.consoles = set([])
+        self.handlertasks = set()
 
     async def add(self, requestid, rsp):
         await self.wshandler(messages.AsyncMessage((requestid, rsp)))
@@ -107,7 +108,11 @@ async def run_handler(hdlr, req):
         raise exc.InvalidArgumentException(
                 'Invalid Session ID or missing request id')
     cloop = asyncio.get_running_loop()
-    cloop.create_task(asyncsession.run_handler(hdlr, requestid))
+    # The loop only keeps a weak reference, so the session holds the task until
+    # it finishes, otherwise the request can be collected while still pending
+    hdlrtask = cloop.create_task(asyncsession.run_handler(hdlr, requestid))
+    asyncsession.handlertasks.add(hdlrtask)
+    hdlrtask.add_done_callback(asyncsession.handlertasks.discard)
     return requestid
 
 

ruff.toml

@@ -76,6 +76,9 @@ select = [
     "B023",     # closure captures a loop variable, so every closure sees the
                 # last value rather than the one from its iteration
     "B035",     # dict comprehension with a static key
+    "RUF006",   # asyncio.create_task result discarded.  The loop keeps only a
+                # weak reference, so an unreferenced task can be collected
+                # while it is still pending, and the work is silently dropped
 ]
 
 # Deliberately not selected, though currently at zero: B905 (zip without an