From dc34aa4f0b90de8ec957d985862b77c849de9fd4 Mon Sep 17 00:00:00 2001 From: Alexandre Perez Date: Thu, 23 Jul 2026 13:12:25 -0700 Subject: [PATCH 1/2] Fix race between adapter connection publication and session attachment `Connection.__init__` publishes a new connection to `_connections` under `_lock`, then releases the lock before classifying it. In that window another thread can attach the connection to a session (via an `attach` request or the launch flow), which sets `conn.server` and the session's `pid`. When the constructor resumes, `sessions.get(self.pid)` returns that just-attached session, and because it is not a process replacement the old code logged "is not expecting replacement" and closed the channel, dropping a session that was already live. This shows up with subprocess debugging and in-place process replacement. Fix it by re-checking under `_lock` whether the connection was already attached (`self.server is not None`) before treating it as an unexpected replacement. Closing the channel under `_lock` is deadlock-free: `JsonMessageChannel.close()` only takes the channel's own lock and the disconnect handler runs on the parser thread. Add a regression test that overlaps publication and attachment. --- src/debugpy/adapter/servers.py | 20 +++- tests/debugpy/adapter/test_servers.py | 159 ++++++++++++++++++++++++++ 2 files changed, 174 insertions(+), 5 deletions(-) create mode 100644 tests/debugpy/adapter/test_servers.py diff --git a/src/debugpy/adapter/servers.py b/src/debugpy/adapter/servers.py index 0aede938..a4eb1f5c 100644 --- a/src/debugpy/adapter/servers.py +++ b/src/debugpy/adapter/servers.py @@ -126,11 +126,21 @@ def __init__(self, sock): log.info("No active debug session for parent process of {0}.", self) else: if self.pid == parent_session.pid: - parent_server = parent_session.server - if not (parent_server and parent_server.connection.process_replaced): - log.error("{0} is not expecting replacement.", parent_session) - self.channel.close() - return + with _lock: + # The connection can be attached to a session as soon as it + # is published above, before this constructor gets a chance to + # classify it. If that already happened, the connection is + # established and must not be treated as an unexpected + # replacement. + if self.server is not None: + return + parent_server = parent_session.server + if not ( + parent_server and parent_server.connection.process_replaced + ): + log.error("{0} is not expecting replacement.", parent_session) + self.channel.close() + return try: parent_session.client.notify_of_subprocess(self) return diff --git a/tests/debugpy/adapter/test_servers.py b/tests/debugpy/adapter/test_servers.py new file mode 100644 index 00000000..aa8e2067 --- /dev/null +++ b/tests/debugpy/adapter/test_servers.py @@ -0,0 +1,159 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See LICENSE in the project root +# for license information. + +import threading + +from debugpy import adapter +from debugpy.adapter import servers, sessions + + +class FakeMessage: + def __init__(self, values): + self.values = values + + def __call__(self, name, _validator, optional=False): + if optional and name not in self.values: + return () + return self.values[name] + + +class FakeStream: + name = "fake stream" + + +class FakeChannel: + def __init__(self, stream, handlers, pid): + self.stream = stream + self.handlers = handlers + self.pid = pid + self.name = "fake channel" + self.closed = False + + def start(self): + pass + + def request(self, command, arguments=None): + assert command == "pydevdSystemInfo" + return FakeMessage( + { + "process": FakeMessage( + { + "pid": self.pid, + } + ) + } + ) + + def close(self): + self.closed = True + + +def test_connection_waits_for_actual_session_attachment(monkeypatch): + # Regression test for a race in Connection.__init__: once a connection is + # published to servers._connections, another thread can attach it to a + # session before the constructor classifies it. The constructor must not + # then close the channel as an "unexpected replacement". Without the fix in + # servers.py, this test fails (the channel is closed / the constructor never + # reaches the post-attachment check). + pid = 1234 + stream = FakeStream() + channel = FakeChannel(stream, None, pid) + attached_session = sessions.Session() + stale_session = sessions.Session() + stale_session.pid = pid + published = threading.Event() + attachment_started = threading.Event() + release_attachment = threading.Event() + classification_waiting = threading.Event() + connection_finished = threading.Event() + connections = [] + attach_errors = [] + connection_errors = [] + original_server = servers.Server + + class ObservedLock: + def __init__(self): + self._lock = threading.RLock() + + def __enter__(self): + if published.is_set() and threading.current_thread().name == "connection": + classification_waiting.set() + self._lock.acquire() + return self + + def __exit__(self, *_): + self._lock.release() + + class ConnectionPublished: + def set(self): + published.set() + + def blocking_server(session, connection): + attachment_started.set() + assert release_attachment.wait(5) + return original_server(session, connection) + + def attach_published_connection(): + try: + assert published.wait(5) + with servers._lock: + connection = servers._connections[-1] + connection.attach_to_session(attached_session) + except Exception as exc: + attach_errors.append(exc) + + def create_connection(): + try: + connections.append(servers.Connection(object())) + except Exception as exc: + connection_errors.append(exc) + finally: + connection_finished.set() + + def get_session(candidate_pid): + if candidate_pid != pid: + return None + assert attachment_started.wait(5) + return stale_session + + monkeypatch.setattr(adapter, "access_token", None) + monkeypatch.setattr(servers, "access_token", None) + monkeypatch.setattr(servers, "_lock", ObservedLock()) + monkeypatch.setattr(servers, "_connections", []) + monkeypatch.setattr(servers, "_connections_changed", ConnectionPublished()) + monkeypatch.setattr(servers, "Server", blocking_server) + monkeypatch.setattr( + servers.messaging.JsonIOStream, "from_socket", lambda *_: stream + ) + monkeypatch.setattr( + servers.messaging, + "JsonMessageChannel", + lambda *_: channel, + ) + monkeypatch.setattr(sessions, "get", get_session) + + attach_thread = threading.Thread(target=attach_published_connection, daemon=True) + connection_thread = threading.Thread( + target=create_connection, + name="connection", + daemon=True, + ) + attach_thread.start() + connection_thread.start() + try: + assert attachment_started.wait(5) + assert classification_waiting.wait(5) + assert not connection_finished.is_set() + finally: + release_attachment.set() + attach_thread.join(5) + connection_thread.join(5) + + assert not attach_thread.is_alive() + assert not connection_thread.is_alive() + assert not attach_errors + assert not connection_errors + assert len(connections) == 1 + assert connections[0].server is attached_session.server + assert not channel.closed From aa823edda1a0ab03014773469351e36ae303ad96 Mon Sep 17 00:00:00 2001 From: Alexandre Perez Date: Thu, 23 Jul 2026 16:03:49 -0700 Subject: [PATCH 2/2] Make the adapter race regression test behavioral --- tests/debugpy/adapter/test_servers.py | 118 +++++++------------------- 1 file changed, 31 insertions(+), 87 deletions(-) diff --git a/tests/debugpy/adapter/test_servers.py b/tests/debugpy/adapter/test_servers.py index aa8e2067..5b5a9d4b 100644 --- a/tests/debugpy/adapter/test_servers.py +++ b/tests/debugpy/adapter/test_servers.py @@ -23,9 +23,9 @@ class FakeStream: class FakeChannel: - def __init__(self, stream, handlers, pid): + def __init__(self, stream, pid): self.stream = stream - self.handlers = handlers + self.handlers = None self.pid = pid self.name = "fake channel" self.closed = False @@ -35,125 +35,69 @@ def start(self): def request(self, command, arguments=None): assert command == "pydevdSystemInfo" - return FakeMessage( - { - "process": FakeMessage( - { - "pid": self.pid, - } - ) - } - ) + return FakeMessage({"process": FakeMessage({"pid": self.pid})}) def close(self): self.closed = True def test_connection_waits_for_actual_session_attachment(monkeypatch): - # Regression test for a race in Connection.__init__: once a connection is - # published to servers._connections, another thread can attach it to a - # session before the constructor classifies it. The constructor must not - # then close the channel as an "unexpected replacement". Without the fix in - # servers.py, this test fails (the channel is closed / the constructor never - # reaches the post-attachment check). + # Attaching a connection stamps session.pid with the connection's pid, so a + # connection attached in the window after publication but before __init__ + # classifies it by pid must survive classification rather than be closed. pid = 1234 stream = FakeStream() - channel = FakeChannel(stream, None, pid) - attached_session = sessions.Session() - stale_session = sessions.Session() - stale_session.pid = pid - published = threading.Event() - attachment_started = threading.Event() - release_attachment = threading.Event() - classification_waiting = threading.Event() - connection_finished = threading.Event() + channel = FakeChannel(stream, pid) + session = sessions.Session() + attachment_complete = threading.Event() connections = [] - attach_errors = [] - connection_errors = [] - original_server = servers.Server + errors = [] - class ObservedLock: - def __init__(self): - self._lock = threading.RLock() - - def __enter__(self): - if published.is_set() and threading.current_thread().name == "connection": - classification_waiting.set() - self._lock.acquire() - return self - - def __exit__(self, *_): - self._lock.release() - - class ConnectionPublished: - def set(self): - published.set() - - def blocking_server(session, connection): - attachment_started.set() - assert release_attachment.wait(5) - return original_server(session, connection) - - def attach_published_connection(): + def attach_like_a_client(): + # Mirror the real client attach path. try: - assert published.wait(5) - with servers._lock: - connection = servers._connections[-1] - connection.attach_to_session(attached_session) + conn = servers.wait_for_connection( + session, lambda c: c.pid == pid, timeout=None + ) + conn.attach_to_session(session) except Exception as exc: - attach_errors.append(exc) + errors.append(exc) + finally: + attachment_complete.set() def create_connection(): try: connections.append(servers.Connection(object())) except Exception as exc: - connection_errors.append(exc) - finally: - connection_finished.set() + errors.append(exc) def get_session(candidate_pid): if candidate_pid != pid: return None - assert attachment_started.wait(5) - return stale_session + # Block classification until the attach completes, forcing the race. + assert attachment_complete.wait(5) + return session monkeypatch.setattr(adapter, "access_token", None) monkeypatch.setattr(servers, "access_token", None) - monkeypatch.setattr(servers, "_lock", ObservedLock()) monkeypatch.setattr(servers, "_connections", []) - monkeypatch.setattr(servers, "_connections_changed", ConnectionPublished()) - monkeypatch.setattr(servers, "Server", blocking_server) + monkeypatch.setattr(servers, "_connections_changed", threading.Event()) monkeypatch.setattr( servers.messaging.JsonIOStream, "from_socket", lambda *_: stream ) - monkeypatch.setattr( - servers.messaging, - "JsonMessageChannel", - lambda *_: channel, - ) + monkeypatch.setattr(servers.messaging, "JsonMessageChannel", lambda *_: channel) monkeypatch.setattr(sessions, "get", get_session) - attach_thread = threading.Thread(target=attach_published_connection, daemon=True) - connection_thread = threading.Thread( - target=create_connection, - name="connection", - daemon=True, - ) + attach_thread = threading.Thread(target=attach_like_a_client, daemon=True) + connection_thread = threading.Thread(target=create_connection, daemon=True) attach_thread.start() connection_thread.start() - try: - assert attachment_started.wait(5) - assert classification_waiting.wait(5) - assert not connection_finished.is_set() - finally: - release_attachment.set() - attach_thread.join(5) - connection_thread.join(5) + attach_thread.join(5) + connection_thread.join(5) assert not attach_thread.is_alive() assert not connection_thread.is_alive() - assert not attach_errors - assert not connection_errors + assert not errors assert len(connections) == 1 - assert connections[0].server is attached_session.server + assert connections[0].server is session.server assert not channel.closed