From 95d6c96363d902e93d8c114baa618ffb6be491d8 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 06:28:41 +0200 Subject: [PATCH 1/3] Relay console keystrokes from a single ordered consumer nodeconsole spawned a task per input byte from the stdin reader callback and kept no reference to it. Two of those tasks overlap as soon as one parks in relay_keypresses waiting on the VNC connection, so keystrokes can reach the node out of order and the escape sequence state (buffer, inputcontext, modkeys) is mutated by more than one task at a time. With the first keystroke relaying slowly, typing abcdef arrives as bcdefa. Those tasks were also unreferenced, which asyncio documents as collectable while still pending, so a keypress could be dropped. Queue the bytes in the reader callback and process them from one long-lived task instead. Keep a reference to the watch_input task as well, since collecting that one takes the whole input handler with it. --- confluent_client/bin/nodeconsole | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/confluent_client/bin/nodeconsole b/confluent_client/bin/nodeconsole index da274319..ab360d84 100755 --- a/confluent_client/bin/nodeconsole +++ b/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'): From 9e41fdc5985ce0dd787bc5c3cfa3b1ac74241500 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 23:18:24 +0200 Subject: [PATCH 2/3] Hold the async HTTP handler task until it finishes run_handler scheduled the coroutine that serves an async HTTP request and dropped the returned task. The event loop only keeps a weak reference, so the task could be collected while still pending, leaving the request unanswered and "Task was destroyed but it is pending!" in the log. The session already outlives the request in _asyncsessions, so it holds the task in a set and discards it from a done callback. --- confluent_server/confluent/asynchttp.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/confluent_server/confluent/asynchttp.py b/confluent_server/confluent/asynchttp.py index 76458f8d..b24b5505 100644 --- a/confluent_server/confluent/asynchttp.py +++ b/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 From cb91589363ce1d2ce824b7cd87eb7f50ba4da1ad Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 23:18:40 +0200 Subject: [PATCH 3/3] Enforce that spawned asyncio tasks are kept referenced RUF006 catches a create_task whose result is discarded. The loop holds only a weak reference, so such a task can be collected while still pending and the work disappears without a trace. Selected last, once the three existing offenders are gone, so the tree stays clean under it from this commit on. --- ruff.toml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/ruff.toml b/ruff.toml index bf49679c..a051d317 100644 --- a/ruff.toml +++ b/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