From 6a349a16e22ed67472d5305201f5a6ab4860a270 Mon Sep 17 00:00:00 2001 From: Antonio Ojea Date: Wed, 30 Sep 2026 10:51:40 +0000 Subject: [PATCH 1/3] deploy: open the VM router's ports in the COS host firewall The GCE VM router sam-router--vm has been a black hole since it was added: Container-Optimized OS drops inbound traffic on the host by default (INPUT policy DROP, ssh excepted), so the VPC rule allow-sam-hub-p2p opened tcp/udp 4501 to a host that never let a SYN through. The router enrolled, so the control plane listed it and every member dialed it on join, paying a dial timeout for nothing: 5s for a sam-node, 10s for the JS SDK, 15s for the Python SDK, which paid it again on every connect() to a peer none of the reachable routers relayed for. Verified on bananas: with the two iptables rules added by hand the port answered at once, and a Python caller's join went from 22s to 8s. The startup script runs on every boot, so the rules do not need to persist. --- .github/workflows/deploy.yaml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.github/workflows/deploy.yaml b/.github/workflows/deploy.yaml index 63c3d584..d20efc51 100644 --- a/.github/workflows/deploy.yaml +++ b/.github/workflows/deploy.yaml @@ -597,6 +597,9 @@ jobs: --metadata=startup-script='#!/bin/bash set -euo pipefail EXTERNAL_IP=$(curl -s -H "Metadata-Flavor: Google" http://metadata.google.internal/computeMetadata/v1/instance/network-interfaces/0/access-configs/0/external-ip) + # COS drops inbound traffic on the host by default; the VPC rule alone does not open the port. + iptables -w -I INPUT -p tcp --dport 4501 -j ACCEPT + iptables -w -I INPUT -p udp --dport 4501 -j ACCEPT docker run -d \ --name sam-router \ --network host \ From a68a9103dbff708ebcbddf67f475819a52c8276d Mon Sep 17 00:00:00 2001 From: Antonio Ojea Date: Wed, 30 Sep 2026 10:51:41 +0000 Subject: [PATCH 2/3] sdk/python: a replaced peer costs connect one dial timeout per step Measured in the bananas canary namespace with agent_mesh at DEBUG: one provider record naming a pod a rollout had replaced cost connect() 45s. Its address answered nothing (15s); the three admitted routers refused the circuit at once; the DHT named the same address, which was dialed again (15s); and the router the control plane lists that the member had not joined through was admitted on its own (15s). Routers keep a provider record for 48 hours by default and a deploy replaces both everything pods, so a caller meets several such records, and the sam-sdk-python probe ran out its 120s budget on them. connect() now dials the peer's own addresses and the circuit through every admitted router at once, then, for a peer none of them reached, every path the DHT and the control plane add, at once: the DHT's addresses this caller has not dialed yet, and the routers not joined through, each admitted on the way. A dead peer costs two dial timeouts, one per step, however many routers there are. The test pins the 45s case at 30s with the same inputs. --- sdk/python/src/agent_mesh/session.py | 119 ++++++++++++++------------- sdk/python/tests/test_session.py | 82 +++++++++++++++++- 2 files changed, 143 insertions(+), 58 deletions(-) diff --git a/sdk/python/src/agent_mesh/session.py b/sdk/python/src/agent_mesh/session.py index f8da6f6c..4c8a1197 100644 --- a/sdk/python/src/agent_mesh/session.py +++ b/sdk/python/src/agent_mesh/session.py @@ -20,7 +20,7 @@ from contextlib import asynccontextmanager from dataclasses import dataclass, field, replace from datetime import datetime -from typing import TYPE_CHECKING, Any, AsyncIterator, Mapping, Optional, Sequence, Union +from typing import TYPE_CHECKING, Any, AsyncIterator, Awaitable, Callable, Mapping, Optional, Sequence, Union import multiaddr import trio @@ -199,15 +199,16 @@ def relay_addresses(self) -> list[str]: async def connect(self, peer: Peer) -> ID: """Connects to a peer, see `Peer`, and returns its peer ID. A banned - peer is refused. A peer named by ID that no admitted router relays for - is looked up in the routers' DHT, and failing that tried through every - 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. 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.""" + peer is refused. The peer's own addresses and the relayed path through + every admitted router are dialed at once. A peer none of them reaches + is looked up in the routers' DHT and tried through every router the + control plane lists that this member has not joined through, again at + once; a relay opens a circuit only for a source it authenticated, so + each such router is admitted on the way. Which router each side joined + through does not decide whether they can talk. A peer a rollout + replaced is still in the routers' tables for a while; its dead + address answers nothing, and costs the caller one dial timeout per + step, not one per address or 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): @@ -227,44 +228,57 @@ async def connect(self, peer: Peer) -> ID: direct.extend(multiaddr.Multiaddr(str(m).removesuffix(suffix)) for m in await dial_addrs(a)) except Exception as err: # noqa: BLE001 - an address this host cannot use; the others are tried failures.append(f"{a}: {err}") - if direct: - try: - await dial(self.host, PeerInfo(target, direct)) - 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}") - if await self._connect_through(target, [multiaddr.Multiaddr(f"{r.addr}/p2p-circuit{suffix}") for r in self.routers], failures): + circuits = [multiaddr.Multiaddr(f"{r.addr}/p2p-circuit{suffix}") for r in self.routers] + if await self._connect_through(target, direct, circuits, [], failures): + return target + tried = {str(a) for a in direct} + routed, relays = await self._routed_addresses(target) + routed = [a for a in routed if str(a) not in tried] + for addr in self._unjoined_routers(target): + if not any(str(r) == str(addr) for r in relays): + relays.append(addr) + if await self._connect_through(target, routed, [], relays, failures): return target - for more in (self._routed_addresses, self._unjoined_router_addresses): - 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.""" + async def _connect_through( + self, target: ID, direct: list[multiaddr.Multiaddr], circuits: list[multiaddr.Multiaddr], relays: list[multiaddr.Multiaddr], failures: list[str] + ) -> bool: + """Dials target every way given at once: its direct addresses as one + dial, each circuit, and each relay after admitting it. The first that + reaches target ends the others, and each that failed adds its reason + to failures.""" reached = False + suffix = f"/p2p/{target}" - async def attempt(ma: multiaddr.Multiaddr, nursery: trio.Nursery) -> None: + async def attempt(what: str, go: Callable[[], Awaitable[object]], nursery: trio.Nursery) -> None: nonlocal reached try: - await self._connect_addr(ma) + await go() except Exception as err: # noqa: BLE001 - the other attempts go on - failures.append(f"{ma}: {err}") + failures.append(f"{what}: {err}") return reached = True nursery.cancel_scope.cancel() + async def through_relay(addr: multiaddr.Multiaddr) -> None: + admitted = await self._admit_router(addr) + await self._connect_addr(multiaddr.Multiaddr(f"{admitted.addr}/p2p-circuit{suffix}")) + async with trio.open_nursery() as nursery: - for ma in addrs: - nursery.start_soon(attempt, ma, nursery) + if direct: + nursery.start_soon(attempt, f"direct {[str(a) for a in direct]}", lambda: dial(self.host, PeerInfo(target, direct)), nursery) + for ma in circuits: + nursery.start_soon(attempt, str(ma), lambda ma=ma: self._connect_addr(ma), nursery) + for addr in relays: + nursery.start_soon(attempt, f"{addr}/p2p-circuit{suffix}", lambda addr=addr: through_relay(addr), 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. - The list is the one the credential carries, refreshed by every - control plane pull, so a router that came up after join is tried too.""" + def _unjoined_routers(self, target: ID) -> list[multiaddr.Multiaddr]: + """The routers the control plane lists that this member has not joined + through. The list is the one the credential carries, refreshed by + every control plane pull, so a router that came up after join is + among them.""" out: list[multiaddr.Multiaddr] = [] for text in self.mesh.credential.router_addresses: try: @@ -274,42 +288,35 @@ async def _unjoined_router_addresses(self, target: ID) -> list[multiaddr.Multiad continue if router == target or str(router) in self.banned or any(r.peer_id == str(router) for r in self.routers): continue - try: - admitted = await self._admit_router(addr) - except Exception as err: # noqa: BLE001 - a router that does not admit us is not a path - logger.debug("router %s did not admit us: %s", router, err) - continue - out.append(multiaddr.Multiaddr(f"{admitted.addr}/p2p-circuit/p2p/{target}")) + out.append(addr) return out - 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 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] = [] + async def _routed_addresses(self, target: ID) -> tuple[list[multiaddr.Multiaddr], list[multiaddr.Multiaddr]]: + """What the routers' DHT knows for a peer: its direct addresses this + host can dial, without the peer suffix, and the relays it reserved on + that are not admitted routers of this member; those the caller tried + already.""" + direct: list[multiaddr.Multiaddr] = [] + relays: list[multiaddr.Multiaddr] = [] seeds = [ID.from_base58(r.peer_id) for r in self.routers] + suffix = f"/p2p/{target}" for ma in await find_peer(self.host, target, seeds): text = str(ma) if "/p2p-circuit" not in text: - out.append(ma if text.endswith(f"/p2p/{target}") else multiaddr.Multiaddr(f"{text}/p2p/{target}")) + try: + direct.extend(multiaddr.Multiaddr(str(m).removesuffix(suffix)) for m in await dial_addrs(ma)) + except Exception: # noqa: BLE001 - an address this host cannot use + pass continue relay_addr = multiaddr.Multiaddr(text[: text.index("/p2p-circuit")]) try: 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 or any(r.peer_id == str(relay) for r in self.routers): + if str(relay) in self.banned or any(r.peer_id == str(relay) for r in self.routers) or any(str(r) == str(relay_addr) for r in relays): 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 - out.append(multiaddr.Multiaddr(f"{relay_addr}/p2p-circuit/p2p/{target}")) - return out + relays.append(relay_addr) + return direct, relays async def _admit_router(self, addr: multiaddr.Multiaddr) -> AdmittedRouter: """Dials a router, runs the handshake and, its role verified, adds it diff --git a/sdk/python/tests/test_session.py b/sdk/python/tests/test_session.py index d250b8e9..d7584119 100644 --- a/sdk/python/tests/test_session.py +++ b/sdk/python/tests/test_session.py @@ -43,6 +43,7 @@ from agent_mesh.authorizer import ProviderAuthorizerOptions from agent_mesh.biscuit import ROLE_ROUTER, BiscuitVerificationError from agent_mesh.controlplane import ROLE_NODE +from agent_mesh.discovery import DiscoveredProvider from agent_mesh.httpx_transport import MeshTransport from agent_mesh.identity import Identity from agent_mesh.libp2p_http import HTTP_PROTOCOL, A2AEndpoint, HTTPResponse, ProviderOptions, http_ingress_handler @@ -371,10 +372,10 @@ def relay_of(ma) -> str: return str(ma).split("/p2p/")[1].split("/")[0][-6:] async def nothing(_target): - return [] + return [], [] session._routed_addresses = nothing # type: ignore[method-assign] - session._unjoined_router_addresses = nothing # type: ignore[method-assign] + session._unjoined_routers = lambda _target: [] # type: ignore[method-assign] async def main(): # Every relay waits on a destination that is gone. @@ -418,6 +419,83 @@ async def one_opens(ma): trio.run(main, clock=trio.testing.MockClock(autojump_threshold=0)) +def test_a_replaced_peer_costs_connect_one_dial_timeout_per_step(monkeypatch): + """A provider record can name a pod a rollout replaced: its address + answers nothing, the admitted routers refuse the circuit at once, the DHT + names the same address again, and the control plane lists routers this + member has not joined through, one of them dark. The address and the + routers are dialed at once, the DHT's copy of the address is not dialed + again, and the routers not joined through are admitted at once: two + dial timeouts in all, one per step.""" + from agent_mesh import session as session_module + from agent_mesh.host import DIAL_TIMEOUT + + admitted_id = "12D3KooWG1pA6goegCncqwbZLSr8pnjUZ6JMAAe6SmnHTgUNCk88" + unjoined = { + "12D3KooWGvdRCJLYATauVWfsieF2j3a2wXZoEQJUS2MsvRdDtgLM": "/ip4/10.0.0.2/tcp/4501", + "12D3KooWBTdQ3QQZztZFaxQSTzJx5ZSbpgM8zfs43VYzBXAFkdZm": "/ip4/203.0.113.7/tcp/4501", + } + target = "12D3KooWA4Xop1JaT3MHxwYMkCepYsv4iPVopMXwCz5iHYdBfeSB" + dead = "/ip4/10.84.4.137/tcp/5002" + + class Host: + def get_connected_peers(self): + return [] + + class Credential: + router_addresses = [f"/ip4/10.0.0.1/tcp/4501/p2p/{admitted_id}"] + [f"{a}/p2p/{p}" for p, a in unjoined.items()] + + class Mesh: + credential = Credential() + + session = MeshSession( + mesh=Mesh(), # type: ignore[arg-type] + host=Host(), # type: ignore[arg-type] + routers=[AdmittedRouter(peer_id=admitted_id, addr=multiaddr.Multiaddr(f"/ip4/10.0.0.1/tcp/4501/p2p/{admitted_id}"), credential=None)], # type: ignore[arg-type] + ) + dials: list[list[str]] = [] + admissions: list[tuple[str, float]] = [] + circuits: list[str] = [] + + async def dial_nobody_answers(host, info): + dials.append([str(a) for a in info.addrs]) + await trio.sleep(DIAL_TIMEOUT) + raise ConnectionError(f"no connection to {info.peer_id} within {DIAL_TIMEOUT:g}s") + + async def refused(ma): + circuits.append(str(ma)) + raise RuntimeError("relay refused to connect: PERMISSION_DENIED") + + async def dht_names_the_same_address(_target): + return [multiaddr.Multiaddr(dead)], [] + + async def admit_nobody_answers(addr): + admissions.append((str(addr), trio.current_time())) + await trio.sleep(DIAL_TIMEOUT) + raise ConnectionError(f"no connection to {addr} within {DIAL_TIMEOUT:g}s") + + monkeypatch.setattr(session_module, "dial", dial_nobody_answers) + session._connect_addr = refused # type: ignore[method-assign] + session._routed_addresses = dht_names_the_same_address # type: ignore[method-assign] + session._admit_router = admit_nobody_answers # type: ignore[method-assign] + + async def main(): + started = trio.current_time() + provider = DiscoveredProvider(peer_id=target, addrs=[f"{dead}/p2p/{target}", f"/ip4/10.84.4.137/udp/5001/quic-v1/p2p/{target}"]) + with pytest.raises(ConnectionError) as err: + await session.connect(provider) + assert trio.current_time() - started == pytest.approx(2 * DIAL_TIMEOUT) + # The dead address once, with the admitted router's circuit alongside. + assert dials == [[dead]] + assert circuits == [f"/ip4/10.0.0.1/tcp/4501/p2p/{admitted_id}/p2p-circuit/p2p/{target}"] + # Both routers not joined through, admitted at the same instant. + assert sorted(a for a, _ in admissions) == sorted(f"{a}/p2p/{p}" for p, a in unjoined.items()) + assert len({t for _, t in admissions}) == 1 + assert str(err.value).count("within 15s") == 3 + + 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 From b507ab1bfdedaffb8985735f83efc3d65dc609b6 Mon Sep 17 00:00:00 2001 From: Antonio Ojea Date: Wed, 30 Sep 2026 10:51:42 +0000 Subject: [PATCH 3/3] deploy: routers on bananas keep a provider record for an hour A router keeps a DHT provider record for 48 hours by default, and every bananas deploy replaces both everything pods, so callers of mcp://everything meet the dead pods of the last two days of deploys before a live one, each costing a dial timeout. Nodes reprovide every five minutes; an hour on bananas outlives a dozen missed reprovides and clears a replaced pod within the hour. Hub keeps the library default: it deploys from release tags, rarely. --dht-provider-addr-ttl is rendered into the router StatefulSet and the VM router from one per-environment value in deploy.yaml. --- .github/k8s/sam-router-template.yaml | 3 +++ .github/workflows/deploy.yaml | 12 ++++++++++-- tests/e2e/canary_manifests.bats | 1 + 3 files changed, 14 insertions(+), 2 deletions(-) diff --git a/.github/k8s/sam-router-template.yaml b/.github/k8s/sam-router-template.yaml index 3f916732..2b2aeca6 100644 --- a/.github/k8s/sam-router-template.yaml +++ b/.github/k8s/sam-router-template.yaml @@ -68,6 +68,9 @@ spec: - "--jwt-path=/var/run/secrets/tokens/sam-token" - "--keys-path=/data/router.key" - "--metrics-addr=0.0.0.0:9090" + # How long a provider record outlives its last reprovide (nodes reprovide + # every 5m); 0s is the library's 48h. Set per environment in deploy.yaml. + - "--dht-provider-addr-ttl=${DHT_PROVIDER_ADDR_TTL}" # /healthz answers as soon as the process is up; /readyz once the # router has enrolled and its libp2p host is online. startupProbe: diff --git a/.github/workflows/deploy.yaml b/.github/workflows/deploy.yaml index d20efc51..57bb8637 100644 --- a/.github/workflows/deploy.yaml +++ b/.github/workflows/deploy.yaml @@ -411,6 +411,10 @@ jobs: VAR_GCP_PROJECT_ID: ${{ vars.GCP_PROJECT_ID }} VAR_CLUSTER_REGION: ${{ vars.CLUSTER_REGION }} VAR_CLUSTER_NAME: ${{ vars.CLUSTER_NAME }} + # Bananas deploys several times a day and every deploy replaces the + # canary pods; with the library's 48h a caller meets their dead + # provider records for two days. Hub keeps the default. + VAR_DHT_PROVIDER_ADDR_TTL: ${{ vars.ENV_NAME == 'bananas' && '1h' || '0s' }} run: | print_rollout_diagnostics() { local namespace="$1" @@ -443,9 +447,10 @@ jobs: export GCP_PROJECT_ID="${VAR_GCP_PROJECT_ID}" export CLUSTER_REGION="${VAR_CLUSTER_REGION}" export CLUSTER_NAME="${VAR_CLUSTER_NAME}" + export DHT_PROVIDER_ADDR_TTL="${VAR_DHT_PROVIDER_ADDR_TTL}" envsubst '${ENV_NAME} ${NAMESPACE} ${GCP_PROJECT_ID} ${CLUSTER_NAME} ${CLUSTER_REGION} ${IMAGE_TAG}' < .github/k8s/sam-control-plane-template.yaml | kubectl apply -f - - envsubst '${ENV_NAME} ${NAMESPACE} ${GCP_PROJECT_ID} ${CLUSTER_NAME} ${CLUSTER_REGION} ${IMAGE_TAG}' < .github/k8s/sam-router-template.yaml | kubectl apply -f - + envsubst '${ENV_NAME} ${NAMESPACE} ${GCP_PROJECT_ID} ${CLUSTER_NAME} ${CLUSTER_REGION} ${IMAGE_TAG} ${DHT_PROVIDER_ADDR_TTL}' < .github/k8s/sam-router-template.yaml | kubectl apply -f - envsubst '${ENV_NAME} ${NAMESPACE} ${GCP_PROJECT_ID} ${CLUSTER_NAME} ${CLUSTER_REGION} ${IMAGE_TAG}' < .github/k8s/sam-console-template.yaml | kubectl apply -f - envsubst '${ENV_NAME} ${NAMESPACE} ${GCP_PROJECT_ID}' < .github/k8s/dns-sync-cronjob-template.yaml | kubectl apply -f - # Managed Prometheus ships with GKE; a cluster without it should not block the mesh rollout. @@ -536,6 +541,7 @@ jobs: VAR_IMAGE_TAG: ${{ env.IMAGE_TAG }} VAR_GCP_PROJECT_ID: ${{ vars.GCP_PROJECT_ID }} VAR_CLUSTER_REGION: ${{ vars.CLUSTER_REGION }} + VAR_DHT_PROVIDER_ADDR_TTL: ${{ vars.ENV_NAME == 'bananas' && '1h' || '0s' }} run: | export ENV_NAME="${VAR_ENV_NAME}" export NAMESPACE="sam-${ENV_NAME}" @@ -543,6 +549,7 @@ jobs: export GCP_PROJECT_ID="${VAR_GCP_PROJECT_ID}" export CLUSTER_REGION="${VAR_CLUSTER_REGION}" export ZONE="${CLUSTER_REGION}-a" + export DHT_PROVIDER_ADDR_TTL="${VAR_DHT_PROVIDER_ADDR_TTL}" echo "Retrieving Admin Token to generate Bootstrap Token..." ADMIN_TOKEN=$(kubectl get secret sam-control-plane-secret-${ENV_NAME} -n ${NAMESPACE} -o jsonpath='{.data.admin-token}' | base64 -d) @@ -610,7 +617,8 @@ jobs: --listen=/ip4/0.0.0.0/tcp/4501 \ --listen=/ip4/0.0.0.0/udp/4501/quic-v1 \ --external-addr=/ip4/${EXTERNAL_IP}/tcp/4501 \ - --external-addr=/ip4/${EXTERNAL_IP}/udp/4501/quic-v1' + --external-addr=/ip4/${EXTERNAL_IP}/udp/4501/quic-v1 \ + --dht-provider-addr-ttl='"${DHT_PROVIDER_ADDR_TTL}"'' - name: Deploy SAM Box Canary env: diff --git a/tests/e2e/canary_manifests.bats b/tests/e2e/canary_manifests.bats index eb4bad0b..d17eba7e 100644 --- a/tests/e2e/canary_manifests.bats +++ b/tests/e2e/canary_manifests.bats @@ -24,6 +24,7 @@ load "lib/container_mesh.bash" export ENV_NAME="e2e" export NAMESPACE="default" export CANARY_NAMESPACE="sam-canary-${ENV_NAME}" +export DHT_PROVIDER_ADDR_TTL="0s" setup_file() { if ! command -v kind >/dev/null 2>&1 || ! command -v kubectl >/dev/null 2>&1; then