deploy: the VM router was a black hole; replaced pods cost the Python SDK 45s each

.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:

.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,13 +541,15 @@ 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}"
           export IMAGE_TAG="${VAR_IMAGE_TAG}"
           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)
@@ -597,6 +604,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 \
@@ -607,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:

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

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

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