From 6eb7fddd58d7ce7fbf7401071274f01231a092e1 Mon Sep 17 00:00:00 2001 From: Antonio Ojea Date: Tue, 29 Sep 2026 16:51:43 +0000 Subject: [PATCH 1/2] sdk/python: refresh the credential from one thread at a time The session refreshes the credential in one worker thread and pulls from the control plane, which may refresh as well, in another; both persist the result with the same temporary file name. On the testnet the loser found its temporary file already renamed by the winner: credential refresh failed, retrying in 30s: [Errno 2] No such file or directory: 'identity.key.tmp' -> 'identity.key' The other outcome is worse: the control plane redeems only the last biscuit it issued, so the loser presents a spent one and the refresh fails until the next attempt. A lock on the member serializes refresh, the pull and save; the fake control plane in the tests now redeems only its last biscuit, as the real one does, and eight refreshes at once all land. --- sdk/python/src/agent_mesh/mesh.py | 55 ++++++++++++++++------------ sdk/python/tests/test_mesh.py | 60 ++++++++++++++++++++++++++----- 2 files changed, 85 insertions(+), 30 deletions(-) diff --git a/sdk/python/src/agent_mesh/mesh.py b/sdk/python/src/agent_mesh/mesh.py index a810e326..2f29495c 100644 --- a/sdk/python/src/agent_mesh/mesh.py +++ b/sdk/python/src/agent_mesh/mesh.py @@ -57,6 +57,10 @@ def __init__(self, identity: Identity, control_plane: ControlPlaneClient, creden self.control_plane = control_plane self._credential = credential self._state_dir = state_dir + # refresh() and sync_control_plane() run in worker threads of the + # session's loops; the control plane redeems only the last biscuit it + # issued, and both write the same state files. + self._lock = threading.RLock() @property def peer_id(self) -> str: @@ -173,29 +177,31 @@ def refresh(self) -> MeshCredential: """Trades the current biscuit for a fresh one and persists it. The control plane redeems only the last biscuit it issued, so a lost refresh result means re-enrolling; persisting before returning keeps that rare.""" - result = self.control_plane.refresh(self.identity, self._credential.biscuit) - control_plane_keys = self._credential.control_plane_keys - try: - control_plane_keys = self.control_plane.keys(control_plane_keys) - except Exception: # noqa: BLE001 - a failed /keys sync must not cost the new biscuit - pass - self._credential = replace( - self._credential, - biscuit=result.biscuit, - expiration=result.expiration, - control_plane_keys=control_plane_keys, - issued_under_keys=list(control_plane_keys), - ) - self.save() - return self._credential + with self._lock: + result = self.control_plane.refresh(self.identity, self._credential.biscuit) + control_plane_keys = self._credential.control_plane_keys + try: + control_plane_keys = self.control_plane.keys(control_plane_keys) + except Exception: # noqa: BLE001 - a failed /keys sync must not cost the new biscuit + pass + self._credential = replace( + self._credential, + biscuit=result.biscuit, + expiration=result.expiration, + control_plane_keys=control_plane_keys, + issued_under_keys=list(control_plane_keys), + ) + self.save() + return self._credential def add_trusted_key(self, key: bytes) -> bool: """Adopts a signing key announced by a KEY_ROTATION event, so peers whose credentials the new key signs verify before the next pull confirms it.""" - if any(bytes(k) == bytes(key) for k in self._credential.control_plane_keys): - return False - self._credential = replace(self._credential, control_plane_keys=[*self._credential.control_plane_keys, bytes(key)]) - return True + with self._lock: + if any(bytes(k) == bytes(key) for k in self._credential.control_plane_keys): + return False + self._credential = replace(self._credential, control_plane_keys=[*self._credential.control_plane_keys, bytes(key)]) + return True def sync_control_plane(self) -> ControlPlaneSync: """The member's pull from the control plane, as sam-node's SyncControlPlane: @@ -205,6 +211,10 @@ def sync_control_plane(self) -> ControlPlaneSync: addresses and the ban set. Each part is attempted even when another fails; the errors are reported together. Gossip events only bring this forward; they are never the only way state arrives.""" + with self._lock: + return self._sync_control_plane() + + def _sync_control_plane(self) -> ControlPlaneSync: errors: list[str] = [] keys_changed = False refreshed = False @@ -255,9 +265,10 @@ def save(self) -> None: """Writes identity and credential to the state directory, if one is configured.""" if self._state_dir is None: return - self._state_dir.mkdir(parents=True, exist_ok=True, mode=0o700) - _write_atomic(self._state_dir / _IDENTITY_FILE, self.identity.to_libp2p_private_key()) - _write_atomic(self._state_dir / _CREDENTIAL_FILE, self._credential.to_json().encode()) + with self._lock: + self._state_dir.mkdir(parents=True, exist_ok=True, mode=0o700) + _write_atomic(self._state_dir / _IDENTITY_FILE, self.identity.to_libp2p_private_key()) + _write_atomic(self._state_dir / _CREDENTIAL_FILE, self._credential.to_json().encode()) def _load_identity(state: Optional[Path]) -> Optional[Identity]: diff --git a/sdk/python/tests/test_mesh.py b/sdk/python/tests/test_mesh.py index 0af392d1..a00a69c1 100644 --- a/sdk/python/tests/test_mesh.py +++ b/sdk/python/tests/test_mesh.py @@ -12,8 +12,10 @@ # See the License for the specific language governing permissions and # limitations under the License. +import base64 import json import stat +import threading import time import urllib.parse from datetime import datetime, timezone @@ -42,11 +44,22 @@ def _ts_s(seconds: int) -> Timestamp: class FakeControlPlane: - """Approves everything and hands out numbered biscuits.""" + """Approves everything and hands out numbered biscuits. With strict_refresh + it redeems only the last biscuit it issued, as the real one does, and + refresh_delay is how long a /refresh takes.""" - def __init__(self, keys_ok=True): + def __init__(self, keys_ok=True, strict_refresh=False, refresh_delay=0.0): self.issued = 0 self.keys_ok = keys_ok + self.strict_refresh = strict_refresh + self.refresh_delay = refresh_delay + self.last_biscuit = b"" + self._lock = threading.Lock() + + def _issue(self, biscuit): + self.issued += 1 + self.last_biscuit = biscuit + return biscuit def _signed_keys(self): ts = int(time.time() * 1000) @@ -58,25 +71,28 @@ def transport(self, method, url, headers, body): if (method, path) == ("POST", "/register"): # The biscuit names the JWT that was presented, so a test can see which. jwt = pb.EnrollRequest.FromString(body).jwt - self.issued += 1 return 200, pb.EnrollResponse( - biscuit_token=f"biscuit-for-{jwt}".encode(), + biscuit_token=self._issue(f"biscuit-for-{jwt}".encode()), control_plane_public_key=CP_KEY.public_key_raw, router_addresses=["/dns4/router.example/tcp/4001/p2p/12D3KooWP8iKhDf3iCMo2H3butNVfdTUtYwYWYQ75jTGnynXPFMp"], expire_time=_ts_s(int(time.time()) + 3600), ).SerializeToString() if (method, path) == ("POST", "/enroll"): - self.issued += 1 return 200, pb.BootstrapEnrollResponse( status=pb.ENROLLMENT_STATUS_APPROVED, - biscuit_token=f"biscuit-{self.issued}".encode(), + biscuit_token=self._issue(f"biscuit-{self.issued + 1}".encode()), control_plane_public_key=CP_KEY.public_key_raw, router_addresses=["/dns4/router.example/tcp/4001/p2p/12D3KooWP8iKhDf3iCMo2H3butNVfdTUtYwYWYQ75jTGnynXPFMp"], expire_time=_ts_s(int(time.time()) + 3600), ).SerializeToString() if (method, path) == ("POST", "/refresh"): - self.issued += 1 - return 200, pb.TokenRefreshResponse(biscuit_token=f"biscuit-{self.issued}".encode(), expire_time=_ts_s(int(time.time()) + 7200)).SerializeToString() + time.sleep(self.refresh_delay) + with self._lock: + presented = base64.b64decode(headers["Authorization"].removeprefix("Bearer ")) + if self.strict_refresh and presented != self.last_biscuit: + return 200, pb.TokenRefreshResponse(error_message="biscuit already redeemed").SerializeToString() + biscuit = self._issue(f"biscuit-{self.issued + 1}".encode()) + return 200, pb.TokenRefreshResponse(biscuit_token=biscuit, expire_time=_ts_s(int(time.time()) + 7200)).SerializeToString() if (method, path) == ("GET", "/keys"): return (200, self._signed_keys().SerializeToString()) if self.keys_ok else (500, b"boom") return 404, f"no route for {method} {path}".encode() @@ -153,6 +169,34 @@ def test_enroll_without_state_dir_keeps_enrollment_key_when_keys_fails(): assert mesh.credential.biscuit == b"biscuit-2" +def test_concurrent_refreshes_run_one_after_the_other(tmp_path): + """A session refreshes the credential from one worker thread and pulls + from the control plane, which may refresh too, from another. The control + plane redeems only the last biscuit it issued and both write the same + state files, so two refreshes at once leave the loser with a spent + biscuit, or renaming a temp file the winner already renamed.""" + cp = FakeControlPlane(strict_refresh=True, refresh_delay=0.05) + state = tmp_path / "state" + mesh = AgentMesh.enroll("http://127.0.0.1:1", bootstrap_token="sbt_secret", state_dir=state, transport=cp.transport) + errors = [] + + def refresh(): + try: + mesh.refresh() + except Exception as err: # noqa: BLE001 - collected for the assertion + errors.append(err) + + threads = [threading.Thread(target=refresh) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + assert errors == [] + assert mesh.credential.biscuit == b"biscuit-9" + assert json.loads((state / "credential.json").read_text())["biscuit"] == base64.b64encode(b"biscuit-9").decode() + assert AgentMesh.load(state, transport=cp.transport).credential.biscuit == b"biscuit-9" + + def test_enroll_refuses_ambiguous_credentials(): cp = FakeControlPlane() with pytest.raises(ValueError, match="exactly one of"): From dc7fcc5132a5337061e723e58ffc35bd58fffe20 Mon Sep 17 00:00:00 2001 From: Antonio Ojea Date: Tue, 29 Sep 2026 16:51:44 +0000 Subject: [PATCH 2/2] sdk/python: a peer that is gone costs connect one wait, not one per router After a rollout the routers' tables still name the pods it replaced, and the sam-sdk-python probe on the testnet ran out its 120s budget on the first provider it tried, printing nothing. connect() walked the relayed paths in turn: a relay whose destination is gone answers a CONNECT only after its own 30s timeout, and a dead provider was tried through every router twice, once as an admitted router and once more as the relay the DHT named for it. py-libp2p's limits on a relayed connection's upgrade add another half minute or more when the relay accepted the circuit and the far end never speaks. The routers of each step are now dialed at once, as join() and the DHT walk already do, and the first circuit that opens ends the others; the DHT step skips relays that are admitted routers, which the first step dialed. A relayed dial is held to DIAL_TIMEOUT as a direct one is, and a circuit given up on is reset so the relay drops it too. The JS SDK dials every address of a peer at once already. A unit test pins the race with three routers that all wait on a dead destination, and one that opens. The SDK mesh integration test gains a relay that accepts every circuit and forwards nothing: every SDK member gives up within its dial timeout and the relay sees each circuit let go. --- sdk/python/src/agent_mesh/relay.py | 23 +++++++-- sdk/python/src/agent_mesh/session.py | 62 ++++++++++++++--------- sdk/python/tests/test_relay.py | 70 ++++++++++++++++++++++++++ sdk/python/tests/test_session.py | 75 +++++++++++++++++++++++++++- tests/integration/sdk_mesh_test.go | 74 ++++++++++++++++++++++++++- 5 files changed, 276 insertions(+), 28 deletions(-) create mode 100644 sdk/python/tests/test_relay.py diff --git a/sdk/python/src/agent_mesh/relay.py b/sdk/python/src/agent_mesh/relay.py index ca3d2b5d..b26dd443 100644 --- a/sdk/python/src/agent_mesh/relay.py +++ b/sdk/python/src/agent_mesh/relay.py @@ -36,7 +36,7 @@ from libp2p.utils.varint import encode_varint_prefixed, read_varint_prefixed_bytes from ._proto import circuit_pb2 as circuit -from .host import open_stream +from .host import DIAL_TIMEOUT, HANGUP_GRACE, open_stream from .identity import canonical_peer_id logger = logging.getLogger("agent_mesh") @@ -82,7 +82,11 @@ async def reserve_relay(host: IHost, relay_peer_id: ID) -> circuit.Reservation: async def dial_through_relay(host: IHost, relay_peer_id: ID, target: ID) -> INetConn: """Opens a connection to `target` through a relay we are connected to, and - registers it with the host so streams can be opened on it.""" + registers it with the host so streams can be opened on it. The relay's + answer and the handshake with the target that follows are each bounded: + py-libp2p's own limits on that upgrade add up to about a minute when the + relay accepted the circuit and the far end never speaks, and a relayed + dial is held to DIAL_TIMEOUT as a direct one is.""" stream = await open_stream(host, relay_peer_id, HOP_PROTOCOL, RELAY_MESSAGE_TIMEOUT) try: with trio.fail_after(RELAY_MESSAGE_TIMEOUT): @@ -99,7 +103,20 @@ async def dial_through_relay(host: IHost, relay_peer_id: ID, target: ID) -> INet # From here the stream is the wire; the usual TLS + yamux upgrade runs on it. circuit_addr = multiaddr.Multiaddr(f"/p2p/{relay_peer_id}/p2p-circuit/p2p/{target}") raw = RawConnection(stream=stream, initiator=True, connection_type=ConnectionType.RELAYED, addresses=[circuit_addr]) - return await host.upgrade_outbound_connection(raw, target) + try: + with trio.fail_after(DIAL_TIMEOUT): + return await host.upgrade_outbound_connection(raw, target) + except BaseException as err: + # A circuit given up on, timed out, failed or cancelled for another + # path that won, is reset so the relay drops it too. + with trio.CancelScope(shield=True), trio.move_on_after(HANGUP_GRACE): + try: + await stream.reset() + except Exception: # noqa: BLE001 - already gone + pass + if isinstance(err, trio.TooSlowError): + raise ConnectionError(f"no secure connection to {target} through relay {relay_peer_id} within {DIAL_TIMEOUT:g}s") from None + raise def stop_stream_handler(host: IHost) -> Callable[[INetStream], object]: diff --git a/sdk/python/src/agent_mesh/session.py b/sdk/python/src/agent_mesh/session.py index 6d1ef571..f8da6f6c 100644 --- a/sdk/python/src/agent_mesh/session.py +++ b/sdk/python/src/agent_mesh/session.py @@ -204,7 +204,10 @@ async def connect(self, peer: Peer) -> ID: router the control plane lists that this member has not joined through; a relay opens a circuit only for a source it authenticated, so each such router is admitted first. Which router each side joined - through does not decide whether they can talk.""" + through does not decide whether they can talk. The routers of each + step are dialed at once: a relay whose destination is gone answers + only after its own timeout, and a peer a rollout replaced must cost + the caller one such wait, not one per router.""" if isinstance(peer, multiaddr.Multiaddr) or (isinstance(peer, str) and peer.startswith("/")): return await self._connect_addr(multiaddr.Multiaddr(str(peer))) if isinstance(peer, str): @@ -230,21 +233,33 @@ async def connect(self, peer: Peer) -> ID: return target except Exception as err: # noqa: BLE001 - the relayed path is tried next failures.append(f"direct {[str(a) for a in direct]}: {err}") - for r in self.routers: - try: - await self._connect_addr(multiaddr.Multiaddr(f"{r.addr}/p2p-circuit{suffix}")) - return target - except Exception as err: # noqa: BLE001 - the next router is tried - failures.append(f"via router {r.peer_id}: {err}") + if await self._connect_through(target, [multiaddr.Multiaddr(f"{r.addr}/p2p-circuit{suffix}") for r in self.routers], failures): + return target for more in (self._routed_addresses, self._unjoined_router_addresses): - for ma in await more(target): - try: - await self._connect_addr(ma) - return target - except Exception as err: # noqa: BLE001 - the next address is tried - failures.append(f"{ma}: {err}") + if await self._connect_through(target, await more(target), failures): + return target raise ConnectionError(f"cannot reach {target}:\n " + "\n ".join(failures)) + async def _connect_through(self, target: ID, addrs: list[multiaddr.Multiaddr], failures: list[str]) -> bool: + """Dials the addresses at once; the first that reaches target ends the + others, and each that failed adds its reason to failures.""" + reached = False + + async def attempt(ma: multiaddr.Multiaddr, nursery: trio.Nursery) -> None: + nonlocal reached + try: + await self._connect_addr(ma) + except Exception as err: # noqa: BLE001 - the other attempts go on + failures.append(f"{ma}: {err}") + return + reached = True + nursery.cancel_scope.cancel() + + async with trio.open_nursery() as nursery: + for ma in addrs: + nursery.start_soon(attempt, ma, nursery) + return reached or target in self.host.get_connected_peers() + async def _unjoined_router_addresses(self, target: ID) -> list[multiaddr.Multiaddr]: """The relayed paths to a peer through the routers the control plane lists that this member has not joined through, each admitted first. @@ -269,9 +284,11 @@ async def _unjoined_router_addresses(self, target: ID) -> list[multiaddr.Multiad async def _routed_addresses(self, target: ID) -> list[multiaddr.Multiaddr]: """The addresses the routers' DHT knows for a peer, relayed ones - through routers this member has admitted by then: a relay opens a - circuit only for a source it authenticated, so a router met this way - is dialed and passed the handshake first, and joins the admitted set.""" + through routers this member has not dialed for it yet: a relay that + is an admitted router was tried by connect() already, and a relay + opens a circuit only for a source it authenticated, so a router met + this way is dialed and passed the handshake first, and joins the + admitted set.""" out: list[multiaddr.Multiaddr] = [] seeds = [ID.from_base58(r.peer_id) for r in self.routers] for ma in await find_peer(self.host, target, seeds): @@ -284,14 +301,13 @@ async def _routed_addresses(self, target: ID) -> list[multiaddr.Multiaddr]: relay = info_from_p2p_addr(relay_addr).peer_id except Exception: # noqa: BLE001 - a circuit address naming no relay is useless continue - if str(relay) in self.banned: + if str(relay) in self.banned or any(r.peer_id == str(relay) for r in self.routers): + continue + try: + await self._admit_router(relay_addr) + except Exception as err: # noqa: BLE001 - a relay that is not a router of this mesh is not used + logger.debug("router %s named by the DHT did not admit us: %s", relay, err) continue - if not any(r.peer_id == str(relay) for r in self.routers): - try: - await self._admit_router(relay_addr) - except Exception as err: # noqa: BLE001 - a relay that is not a router of this mesh is not used - logger.debug("router %s named by the DHT did not admit us: %s", relay, err) - continue out.append(multiaddr.Multiaddr(f"{relay_addr}/p2p-circuit/p2p/{target}")) return out diff --git a/sdk/python/tests/test_relay.py b/sdk/python/tests/test_relay.py new file mode 100644 index 00000000..2d32d9ee --- /dev/null +++ b/sdk/python/tests/test_relay.py @@ -0,0 +1,70 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""What a dial through a relay must do that py-libp2p does not: end at the +SDK's dial timeout when the relay accepted the circuit but the far end never +speaks. py-libp2p's own limits on that upgrade add up to about a minute, +and a caller that walks several relayed paths pays it for each.""" + +import multiaddr +import pytest +import trio +from libp2p.peer.id import ID +from libp2p.peer.peerinfo import info_from_p2p_addr +from libp2p.utils.varint import encode_varint_prefixed, read_varint_prefixed_bytes + +from agent_mesh import relay as relay_module +from agent_mesh._proto import circuit_pb2 as circuit +from agent_mesh.host import create_mesh_host +from agent_mesh.identity import Identity +from agent_mesh.relay import HOP_PROTOCOL, dial_through_relay + +TARGET = ID.from_base58("12D3KooWG1pA6goegCncqwbZLSr8pnjUZ6JMAAe6SmnHTgUNCk88") + + +def test_a_circuit_the_far_end_never_speaks_on_ends_at_the_dial_timeout(monkeypatch): + """A relay that answers CONNECT with OK and then forwards nothing, as one + does while the destination it accepted the circuit for never completes + its side of the handshake. The dial ends at DIAL_TIMEOUT with a + ConnectionError, and the relay sees the circuit let go.""" + monkeypatch.setattr(relay_module, "DIAL_TIMEOUT", 0.5, raising=False) + released = trio.Event() + + async def accept_and_stall(stream): + req = circuit.HopMessage.FromString(await read_varint_prefixed_bytes(stream)) + assert req.type == circuit.HopMessage.CONNECT + assert ID(req.peer.id) == TARGET + await stream.write(encode_varint_prefixed(circuit.HopMessage(type=circuit.HopMessage.STATUS, status=circuit.OK).SerializeToString())) + # Nothing comes from the far end; the caller's reset ends this read. + try: + await stream.read(1) + except Exception: # noqa: BLE001 - the reset is the outcome looked for + pass + released.set() + + async def main(): + relay, relay_listen = create_mesh_host(Identity.generate(), ["/ip4/127.0.0.1/tcp/0"]) + client, _ = create_mesh_host(Identity.generate()) + relay.set_stream_handler(HOP_PROTOCOL, accept_and_stall) + async with relay.run(listen_addrs=relay_listen), client.run(listen_addrs=[]): + await client.connect(info_from_p2p_addr(multiaddr.Multiaddr(f"{relay.get_addrs()[0]}"))) + started = trio.current_time() + with pytest.raises(ConnectionError, match=f"through relay {relay.get_id()} within 0.5s"): + await dial_through_relay(client, relay.get_id(), TARGET) + assert 0.5 <= trio.current_time() - started < 3 + with trio.fail_after(3): + await released.wait() + assert TARGET not in client.get_connected_peers() + + trio.run(main) diff --git a/sdk/python/tests/test_session.py b/sdk/python/tests/test_session.py index fdbe5e8e..d250b8e9 100644 --- a/sdk/python/tests/test_session.py +++ b/sdk/python/tests/test_session.py @@ -29,6 +29,7 @@ from libp2p import new_host from libp2p.crypto.ed25519 import create_new_key_pair from libp2p.custom_types import TProtocol +from libp2p.peer.id import ID from libp2p.peer.peerinfo import info_from_p2p_addr from libp2p.security.tls.transport import PROTOCOL_ID as TLS_PROTOCOL_ID from libp2p.security.tls.transport import TLSTransport @@ -48,7 +49,7 @@ from agent_mesh.mcp_client import LabelsNotSatisfiedError from agent_mesh.mesh import AgentMesh from agent_mesh.relay import HOP_PROTOCOL as RELAY_HOP_PROTOCOL -from agent_mesh.session import MeshSession +from agent_mesh.session import AdmittedRouter, MeshSession from google.protobuf.timestamp_pb2 import Timestamp @@ -345,6 +346,78 @@ async def main(): trio.run(main, clock=trio.testing.MockClock(autojump_threshold=0)) +def test_connect_dials_the_routers_at_once(): + """A relay whose destination a rollout replaced answers a CONNECT only + after its own timeout. The routers are dialed at once, so a dead peer + costs the caller one such wait, not one per router, and the first + circuit that opens ends the other attempts.""" + router_ids = [ + "12D3KooWG1pA6goegCncqwbZLSr8pnjUZ6JMAAe6SmnHTgUNCk88", + "12D3KooWGvdRCJLYATauVWfsieF2j3a2wXZoEQJUS2MsvRdDtgLM", + "12D3KooWBTdQ3QQZztZFaxQSTzJx5ZSbpgM8zfs43VYzBXAFkdZm", + ] + target = "12D3KooWA4Xop1JaT3MHxwYMkCepYsv4iPVopMXwCz5iHYdBfeSB" + routers = [AdmittedRouter(peer_id=p, addr=multiaddr.Multiaddr(f"/ip4/10.0.0.{i}/tcp/4501/p2p/{p}"), credential=None) for i, p in enumerate(router_ids, 1)] # type: ignore[arg-type] + + class Host: + def get_connected_peers(self): + return [] + + session = MeshSession(mesh=None, host=Host(), routers=routers) # type: ignore[arg-type] + dialed: list[str] = [] + ended: list[str] = [] + + def relay_of(ma) -> str: + return str(ma).split("/p2p/")[1].split("/")[0][-6:] + + async def nothing(_target): + return [] + + session._routed_addresses = nothing # type: ignore[method-assign] + session._unjoined_router_addresses = nothing # type: ignore[method-assign] + + async def main(): + # Every relay waits on a destination that is gone. + async def all_wait(ma): + dialed.append(relay_of(ma)) + try: + await trio.sleep(10) + finally: + ended.append(relay_of(ma)) + raise RuntimeError("CONNECTION_FAILED") + + session._connect_addr = all_wait # type: ignore[method-assign] + started = trio.current_time() + with pytest.raises(ConnectionError) as err: + await session.connect(target) + assert trio.current_time() - started == pytest.approx(10) + assert sorted(dialed) == sorted(p[-6:] for p in router_ids) + assert str(err.value).count("CONNECTION_FAILED") == 3 + + # One relays for the peer; its circuit ends the others' waits. + dialed.clear() + ended.clear() + + async def one_opens(ma): + dialed.append(relay_of(ma)) + if relay_of(ma) == router_ids[1][-6:]: + await trio.sleep(1) + return ID.from_base58(target) + try: + await trio.sleep(10) + finally: + ended.append(relay_of(ma)) + raise RuntimeError("CONNECTION_FAILED") + + session._connect_addr = one_opens # type: ignore[method-assign] + started = trio.current_time() + assert await session.connect(target) == ID.from_base58(target) + assert trio.current_time() - started == pytest.approx(1) + assert sorted(ended) == sorted(p[-6:] for p in (router_ids[0], router_ids[2])) + + trio.run(main, clock=trio.testing.MockClock(autojump_threshold=0)) + + def test_a_dropped_router_connection_is_reserved_again_before_the_ttl(): """A router restart takes the reservation with the connection. The member notices within the check interval and reserves again, authenticated diff --git a/tests/integration/sdk_mesh_test.go b/tests/integration/sdk_mesh_test.go index 364ff88b..51c94f05 100644 --- a/tests/integration/sdk_mesh_test.go +++ b/tests/integration/sdk_mesh_test.go @@ -35,6 +35,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "testing" "time" @@ -48,6 +49,9 @@ import ( "github.com/libp2p/go-libp2p/core/host" "github.com/libp2p/go-libp2p/core/network" "github.com/libp2p/go-libp2p/core/peer" + circuitpb "github.com/libp2p/go-libp2p/p2p/protocol/circuitv2/pb" + circuitproto "github.com/libp2p/go-libp2p/p2p/protocol/circuitv2/proto" + circuitutil "github.com/libp2p/go-libp2p/p2p/protocol/circuitv2/util" libp2ptls "github.com/libp2p/go-libp2p/p2p/security/tls" "github.com/libp2p/go-msgio" "github.com/multiformats/go-multiaddr" @@ -528,6 +532,36 @@ func TestNativeSDKsMesh(t *testing.T) { } }) + // A relay that accepted the circuit while nobody speaks on the far end, + // as a caller sees one whose destination never completes its handshake. + // The dial ends at the SDK's dial timeout with an error and the circuit + // is let go, in every SDK. Every member dials at once, so this costs the + // test one timeout, not one per SDK. + t.Run("stalled-relay", func(t *testing.T) { + relayAddr, released := startStalledRelay(t) + stranger := "12D3KooWA4Xop1JaT3MHxwYMkCepYsv4iPVopMXwCz5iHYdBfeSB" + for _, m := range members { + m.write(t, map[string]string{"cmd": "auth", "addr": relayAddr + "/p2p-circuit/p2p/" + stranger}) + } + for _, m := range members { + var res sdkAuthResult + if line := m.readLine(t, 20*time.Second); json.Unmarshal(line, &res) != nil { + t.Fatalf("%s member: auth answered %q", m.name, line) + } + if res.OK { + t.Fatalf("%s reached a peer through a relay that delivered nothing", m.name) + } + t.Logf("%s gave up on the stalled circuit: %s", m.name, res.Error) + } + deadline := time.Now().Add(5 * time.Second) + for released.Load() < int32(len(members)) && time.Now().Before(deadline) { + time.Sleep(100 * time.Millisecond) + } + if got := released.Load(); got != int32(len(members)) { + t.Fatalf("%d of %d stalled circuits were let go; the rest are leaked on the relay", got, len(members)) + } + }) + // Every SDK member is an agent: it accepts A2A requests for a2a://agent, // answered in the runner's process, reachable by peer ID through the // router. It publishes nothing; the policy rules it evaluates are the @@ -933,12 +967,18 @@ func launchSDKMember(t *testing.T, name string, cmd *exec.Cmd, root, baseURL, ad } func (m *sdkMember) send(t *testing.T, command map[string]string) []byte { + t.Helper() + m.write(t, command) + return m.readLine(t, 20*time.Second) +} + +// write sends a command without waiting for its answer; readLine collects it. +func (m *sdkMember) write(t *testing.T, command map[string]string) { t.Helper() line, _ := json.Marshal(command) if _, err := m.stdin.Write(append(line, '\n')); err != nil { t.Fatalf("%s member: write command: %v", m.name, err) } - return m.readLine(t, 20*time.Second) } // authRaw asks the member to connect to addr and run the auth handshake. @@ -1484,6 +1524,38 @@ func refuseForgedFrame(t *testing.T, ctx context.Context, h host.Host, target pe } } +// startStalledRelay is a circuit relay that answers every CONNECT with OK and +// forwards nothing: what a caller sees when the relay's destination accepted +// the circuit but never completes its side of the handshake. It speaks no +// other protocol, so no member admits it as a router; a caller reaches it +// only by an explicit /p2p-circuit address. Returns its address and the +// count of circuits the callers let go of, by resetting or closing them. +func startStalledRelay(t *testing.T) (string, *atomic.Int32) { + t.Helper() + h, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0")) + if err != nil { + t.Fatalf("failed to create the stalled relay: %v", err) + } + t.Cleanup(func() { _ = h.Close() }) + released := new(atomic.Int32) + h.SetStreamHandler(circuitproto.ProtoIDv2Hop, func(s network.Stream) { + defer func() { _ = s.Close() }() + var msg circuitpb.HopMessage + if err := circuitutil.NewDelimitedReader(s, 4096).ReadMsg(&msg); err != nil || msg.GetType() != circuitpb.HopMessage_CONNECT { + return + } + ok := circuitpb.HopMessage{Type: circuitpb.HopMessage_STATUS.Enum(), Status: circuitpb.Status_OK.Enum()} + if err := circuitutil.NewDelimitedWriter(s).WriteMsg(&ok); err != nil { + return + } + // The caller's handshake bytes arrive and nothing answers them; the + // read ends when the caller resets or closes the circuit. + _, _ = io.Copy(io.Discard, s) + released.Add(1) + }) + return h.Addrs()[0].String() + "/p2p/" + h.ID().String(), released +} + // pickDirectAddr is the member's loopback TCP address with its peer ID. func pickDirectAddr(t *testing.T, addrs []string, peerID string) string { t.Helper()