diff --git a/shard_core/service/peer.py b/shard_core/service/peer.py index f0d040b..55dbe80 100644 --- a/shard_core/service/peer.py +++ b/shard_core/service/peer.py @@ -1,7 +1,6 @@ import asyncio import logging -import httpx import requests from fastapi.requests import Request from http_message_signatures import HTTPSignatureKeyResolver, algorithms @@ -30,7 +29,13 @@ async def update_all_peer_pubkeys(): async with db_conn() as conn: all_peers = await db_peers.get_all(conn) peers_with_pubkey = [Peer(**p) for p in all_peers if p.get("public_bytes_b64")] - await asyncio.gather(*[update_peer_meta(peer) for peer in peers_with_pubkey]) + results = await asyncio.gather( + *[update_peer_meta(peer) for peer in peers_with_pubkey], + return_exceptions=True, + ) + for peer, result in zip(peers_with_pubkey, results): + if isinstance(result, Exception): + log.warning(f"Failed to update peer meta for {peer.short_id}: {result}") async def update_peer_meta(peer: Peer): @@ -49,7 +54,7 @@ def do_request(): try: response.raise_for_status() - except httpx.HTTPStatusError as e: + except requests.HTTPError as e: log.debug(f"Could not update peer meta for {peer.short_id}: {e}") async with db_conn() as conn: await db_peers.update_by_id(conn, peer.id, {"is_reachable": False}) diff --git a/tests/test_peer_refresh.py b/tests/test_peer_refresh.py new file mode 100644 index 0000000..2db3c49 --- /dev/null +++ b/tests/test_peer_refresh.py @@ -0,0 +1,95 @@ +import logging + +import responses +from starlette import status + +from shard_core.data_model.identity import Identity, OutputIdentity +from shard_core.database import peers as db_peers +from shard_core.database.connection import db_conn +from shard_core.data_model.peer import Peer +from shard_core.service.peer import update_all_peer_pubkeys + + +def _whoareyou_url(identity: Identity) -> str: + return f"https://{identity.short_id}.freeshard.cloud/core/public/meta/whoareyou" + + +async def _insert_peer(identity: Identity, name: str, is_reachable: bool = True): + async with db_conn() as conn: + await db_peers.insert( + conn, + { + "id": identity.id, + "name": name, + "public_bytes_b64": identity.public_key_pem, + "is_reachable": is_reachable, + }, + ) + + +async def _get_peer(identity: Identity) -> Peer: + async with db_conn() as conn: + return Peer(**await db_peers.get_by_id_prefix(conn, identity.id)) + + +async def test_unreachable_peer_does_not_block_others(db): + bad = Identity.create("bad peer") + good = Identity.create("good peer") + await _insert_peer(bad, name="stale bad") + await _insert_peer(good, name="stale good", is_reachable=False) + + with responses.RequestsMock(assert_all_requests_are_fired=False) as rsps: + rsps.get(_whoareyou_url(bad), status=status.HTTP_500_INTERNAL_SERVER_ERROR) + rsps.get( + _whoareyou_url(good), + json=OutputIdentity(**good.model_dump()).model_dump(), + ) + + await update_all_peer_pubkeys() + + # both peers were really contacted (the 500 path was exercised, not a + # silent ConnectionError fallback from a mismatched URL) + assert len(rsps.calls) == 2 + + good_peer = await _get_peer(good) + assert good_peer.name == "good peer" + assert good_peer.is_reachable is True + + bad_peer = await _get_peer(bad) + assert bad_peer.is_reachable is False + + +async def test_unexpected_peer_error_does_not_block_others(db, memory_logger): + broken = Identity.create("broken peer") + good = Identity.create("good peer") + await _insert_peer(broken, name="stale broken") + await _insert_peer(good, name="stale good", is_reachable=False) + + with responses.RequestsMock(assert_all_requests_are_fired=False) as rsps: + # broken peer answers with a different identity than its stored id, + # which raises inside update_peer_meta and is not caught there + rsps.get( + _whoareyou_url(broken), + json=OutputIdentity( + **Identity.create("impostor").model_dump() + ).model_dump(), + ) + rsps.get( + _whoareyou_url(good), + json=OutputIdentity(**good.model_dump()).model_dump(), + ) + + await update_all_peer_pubkeys() + + good_peer = await _get_peer(good) + assert good_peer.name == "good peer" + assert good_peer.is_reachable is True + + # the unexpected error is swallowed by the gather and left the broken peer + # untouched, and it was surfaced via a warning + broken_peer = await _get_peer(broken) + assert broken_peer.name == "stale broken" + assert any( + r.levelno == logging.WARNING and broken.short_id in r.getMessage() + for r in memory_logger.records + )