titan/atlas-iac
3
0
Fork 0
Code Issues Pull Requests 9 Actions Packages Projects Releases Wiki Activity

hermes: classify mediator conflicts as publication rejections

Browse Source
This commit is contained in:
jenkins 2026-09-13 18:37:56 -05:00
parent 00b8780345
commit 6cfeff74d6
2 changed files with 23 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
services/hermes/scripts/execution_pool_worker.py
View File

@ -36,8 +36,17 @@ def _post(url: str, value: dict[str, Any], timeout: int = 60) -> dict[str, Any]:
url, data=canonical_json(value), method="POST", url, data=canonical_json(value), method="POST",
headers={"Content-Type": "application/json", "Cache-Control": "no-store"}, headers={"Content-Type": "application/json", "Cache-Control": "no-store"},
) )
with urllib.request.urlopen(request, timeout=timeout) as response: try:
body = response.read(64 * 1024 + 1) with urllib.request.urlopen(request, timeout=timeout) as response:
body = response.read(64 * 1024 + 1)
except urllib.error.HTTPError as error:
if error.code == 409:
try:
error.read(64 * 1024 + 1)
except (AttributeError, OSError):
pass
raise ProtocolError("local mediator rejected request") from error
raise
if len(body) > 64 * 1024: if len(body) > 64 * 1024:
raise ProtocolError("server response exceeds the wire limit") raise ProtocolError("server response exceeds the wire limit")
document = json.loads(body) document = json.loads(body)
@ -47,7 +56,7 @@ def _post(url: str, value: dict[str, Any], timeout: int = 60) -> dict[str, Any]:
def _client(operation: str, **values: Any) -> dict[str, Any]: def _client(operation: str, **values: Any) -> dict[str, Any]:
response = _post(f"{CLIENT}/v1/client", {"operation": operation, **values}) response = _post(f"{CLIENT}/v1/client", {"operation": operation, **values})
if response.get("error"): if response.get("error"):
raise ProtocolError(str(response["error"])) raise ProtocolError("local mediator rejected request")
return response return response
def _poll() -> dict[str, Any] | None: def _poll() -> dict[str, Any] | None:
assignment = _client("poll").get("assignment") assignment = _client("poll").get("assignment")

12
testing/tests/test_hermes_execution_pool_worker_v2.py
View File

@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import io
import json import json
import os import os
import subprocess import subprocess
@ -63,9 +64,18 @@ def test_post_client_and_poll_validate_every_response_boundary(monkeypatch):
) )
with pytest.raises(protocol.ProtocolError, match="object"): with pytest.raises(protocol.ProtocolError, match="object"):
worker._post("http://mediator", {}) worker._post("http://mediator", {})
conflict = urllib.error.HTTPError(
"http://mediator", 409, "Conflict", {}, io.BytesIO(b'{"error":"untrusted detail"}')
)
monkeypatch.setattr(
worker.urllib.request, "urlopen", lambda *_a, **_k: (_ for _ in ()).throw(conflict)
)
with pytest.raises(protocol.ProtocolError, match="local mediator rejected") as rejected:
worker._post("http://mediator", {})
assert "untrusted detail" not in str(rejected.value)
monkeypatch.setattr(worker, "_post", lambda *_a, **_k: {"error": "denied"}) monkeypatch.setattr(worker, "_post", lambda *_a, **_k: {"error": "denied"})
with pytest.raises(protocol.ProtocolError, match="denied"): with pytest.raises(protocol.ProtocolError, match="local mediator rejected"):
worker._client("poll") worker._client("poll")
monkeypatch.setattr(worker, "_post", lambda *_a, **_k: {"safe": True}) monkeypatch.setattr(worker, "_post", lambda *_a, **_k: {"safe": True})
assert worker._client("poll") == {"safe": True} assert worker._client("poll") == {"safe": True}