From a4b7b45e139038bc9ab6d6e113a14c28b6b193e7 Mon Sep 17 00:00:00 2001 From: Godopu Date: Sun, 6 Sep 2026 15:44:33 +0900 Subject: [PATCH] feat(issue-5): persistent MQTT publisher prototype and benchmark harness Splits publish_event.main into three pieces with no behaviour change: prepare_event load the job, consume the seq, build the signed payload, decide retain. The seq is burned here, before any transport runs, so a failed publish can never leave a number a replayed packet could reuse. record_outcome audit record, registry event log and status sync -- all three best effort, all three regardless of the publish result. That is exactly the contract G-1/G-2/G-4 pin down, now in one place. main parse, prepare, publish with retry, record. publisher.py holds the two transports behind one interface. OneShotPublisher is the current behaviour. PersistentPublisher keeps a single connection and lets paho reconnect underneath, raising on an unacknowledged publish so the caller's retry policy stays the only thing deciding attempt counts. It refuses an event for a different broker rather than publishing to the wrong place. Nothing in production uses the persistent path: the CLI and shell helpers exit after one publish. The consumer is the phase 2a engine, which pays the CONNACK cost on every event today -- this job's own started event took ~12.5s because of one CONNACK timeout plus reconnect. bench.py measures both arms. The fake transport is deterministic and is what the tests assert on; the real-broker arm is informational, since the PoC default is a public broker whose latency is not ours to control. Verified: N publishes cost N connections one-shot and 1 persistent; the two transports leave field-for-field identical registry and audit records including the HMAC signature and retain flag; an unacknowledged publish retries on the same connection and records published=false; a second broker is refused. No test opens a socket. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd --- .agents/skills/lib_py/delegation/bench.py | 178 ++++++++++++++++++ .../skills/lib_py/delegation/publish_event.py | 137 +++++++++----- .agents/skills/lib_py/delegation/publisher.py | 169 +++++++++++++++++ tests/test_issue5_phase1_delegation.py | 160 ++++++++++++++++ 4 files changed, 600 insertions(+), 44 deletions(-) create mode 100644 .agents/skills/lib_py/delegation/bench.py create mode 100644 .agents/skills/lib_py/delegation/publisher.py diff --git a/.agents/skills/lib_py/delegation/bench.py b/.agents/skills/lib_py/delegation/bench.py new file mode 100644 index 0000000..6f6826f --- /dev/null +++ b/.agents/skills/lib_py/delegation/bench.py @@ -0,0 +1,178 @@ +"""Benchmark harness for the persistent-connection prototype. + +Two transports, deliberately separated: + + --transport fake a stub client with a synthetic connect latency. Fully + deterministic, touches no network, and is what CI asserts + on: N publishes cost N connections one-shot and exactly 1 + persistent. + --transport broker the real broker named by the job record. Informational + only: the PoC default is a public broker whose CONNACK + latency is not ours to control, so no gate depends on it. + +Usage: + python -m lib_py.delegation.bench --n 50 --transport fake [--connect-latency-ms 50] + [--job ] [--registry-dir DIR] [--out FILE] +""" +from __future__ import annotations + +import argparse +import json +import sys +import time +from typing import List, Optional + +from lib_py.delegation import mqtt_common, publish_event +from lib_py.delegation.publisher import OneShotPublisher, PersistentPublisher + + +class FakeInfo(object): + def __init__(self, published=True): + self._published = published + + def wait_for_publish(self, timeout=None): + return None + + def is_published(self): + return self._published + + +class FakeClient(object): + """Minimal paho stand-in: counts connects and charges a synthetic latency.""" + + def __init__(self, connect_latency=0.0, counters=None): + self.connect_latency = connect_latency + self.counters = counters if counters is not None else {} + self.on_connect = None + + def connect(self, host, port, keepalive=60): + if self.connect_latency: + time.sleep(self.connect_latency) + self.counters["connect"] = self.counters.get("connect", 0) + 1 + if self.on_connect: + self.on_connect(self, None, None, 0, None) + + def loop_start(self): + return None + + def loop_stop(self): + return None + + def reconnect_delay_set(self, min_delay=1, max_delay=16): + return None + + def publish(self, topic, payload=None, qos=0, retain=False): + self.counters["publish"] = self.counters.get("publish", 0) + 1 + return FakeInfo() + + def disconnect(self): + return None + + +def fake_factory(connect_latency, counters): + def factory(_role, _config): + return FakeClient(connect_latency=connect_latency, counters=counters) + return factory + + +def _percentile(values: List[float], pct: float) -> float: + if not values: + return 0.0 + ordered = sorted(values) + idx = int(round((pct / 100.0) * (len(ordered) - 1))) + return ordered[idx] + + +def _run(publisher, prepared, n: int) -> dict: + latencies = [] + failures = 0 + for _ in range(n): + start = time.monotonic() + ok, _err = publisher.publish_with_retry(prepared, attempts=1) + latencies.append((time.monotonic() - start) * 1000.0) + if not ok: + failures += 1 + return { + "publishes": n, + "connects": getattr(publisher, "connect_count", None), + "failures": failures, + "latency_ms": { + "p50": round(_percentile(latencies, 50), 3), + "p95": round(_percentile(latencies, 95), 3), + "max": round(max(latencies), 3) if latencies else 0.0, + }, + } + + +def run_benchmark(n=50, transport="fake", connect_latency_ms=50.0, + job_id=None, registry_dir=None) -> dict: + registry_dir = registry_dir or mqtt_common.DEFAULT_REGISTRY_DIR + + if transport == "fake": + prepared = publish_event.PreparedEvent( + job_id="benchjob", event="progress", detail="bench", seq=0, + topic="python/mqtt/jobs/benchjob/events", retain=False, + payload={"seq": 0}, body=b'{"seq": 0}', + config=mqtt_common.broker_config_from_env(), + ) + latency = connect_latency_ms / 1000.0 + one_counters, persist_counters = {}, {} + one = OneShotPublisher(client_factory=fake_factory(latency, one_counters)) + persist = PersistentPublisher(client_factory=fake_factory(latency, persist_counters)) + try: + result = { + "transport": "fake", + "connect_latency_ms": connect_latency_ms, + "one_shot": _run(one, prepared, n), + "persistent": _run(persist, prepared, n), + } + finally: + persist.close() + result["one_shot"]["client_connects"] = one_counters.get("connect", 0) + result["persistent"]["client_connects"] = persist_counters.get("connect", 0) + return result + + if not job_id: + raise ValueError("--transport broker requires --job") + prepared = publish_event.prepare_event( + job_id=job_id, event="progress", detail="bench", registry_dir=registry_dir, + ) + one = OneShotPublisher() + persist = PersistentPublisher() + try: + return { + "transport": "broker", + "broker": "{}:{}".format(prepared.config.host, prepared.config.port), + "one_shot": _run(one, prepared, n), + "persistent": _run(persist, prepared, n), + } + finally: + persist.close() + + +def main(argv: Optional[List[str]] = None) -> int: + mqtt_common.ensure_env_loaded() + parser = argparse.ArgumentParser(prog="lib_py.delegation.bench") + parser.add_argument("--n", type=int, default=50) + parser.add_argument("--transport", default="fake", choices=["fake", "broker"]) + parser.add_argument("--connect-latency-ms", type=float, default=50.0) + parser.add_argument("--job", default=None) + parser.add_argument("--registry-dir", default=None) + parser.add_argument("--out", default=None) + args = parser.parse_args(argv) + + result = run_benchmark( + n=args.n, transport=args.transport, + connect_latency_ms=args.connect_latency_ms, + job_id=args.job, registry_dir=args.registry_dir, + ) + text = json.dumps(result, indent=2, sort_keys=True) + if args.out: + with open(args.out, "w", encoding="utf-8") as fh: + fh.write(text + "\n") + print(text) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.agents/skills/lib_py/delegation/publish_event.py b/.agents/skills/lib_py/delegation/publish_event.py index 269a98b..923f9e8 100755 --- a/.agents/skills/lib_py/delegation/publish_event.py +++ b/.agents/skills/lib_py/delegation/publish_event.py @@ -124,6 +124,85 @@ def _publish_once(config, topic: str, body: bytes, retain: bool) -> None: pass +class PreparedEvent(object): + """Everything decided before the wire: payload, topic, broker, retain flag. + + Splitting this out lets a different transport (one-shot vs persistent) be + swapped in without touching the parts that must not vary: the seq is already + consumed, the HMAC is already computed over the final payload, and the audit + record is written from these exact fields. + """ + + __slots__ = ("job_id", "event", "detail", "seq", "topic", "retain", + "payload", "body", "config") + + def __init__(self, job_id, event, detail, seq, topic, retain, payload, body, config): + self.job_id = job_id + self.event = event + self.detail = detail + self.seq = seq + self.topic = topic + self.retain = retain + self.payload = payload + self.body = body + self.config = config + + +def prepare_event(job_id, event, detail="", data=None, + registry_dir=DEFAULT_REGISTRY_DIR, retained=False) -> PreparedEvent: + """Load the job, consume the next seq, and build the signed payload. + + Consuming the seq here (before any transport runs) is deliberate: a failed + publish must still burn the number, so a replayed packet can never carry a + seq the subscriber would accept. + """ + job = load_job(job_id, registry_dir) + config = broker_config_from_job(job) + topic = job.get("topic_prefix") + topic = f"{topic}/events" if topic else events_topic_for(job_id) + seq = next_seq(job_id, registry_dir) + payload = build_payload( + job_id=job_id, + seq=seq, + event=event, + detail=detail, + data=data, + auth_token=job.get("auth_token"), + ) + body = json.dumps(payload, ensure_ascii=False).encode("utf-8") + retain = retained or event in TERMINAL_EVENTS + return PreparedEvent(job_id, event, detail, seq, topic, retain, payload, body, config) + + +def record_outcome(prepared: PreparedEvent, registry_dir: str, + publish_ok: bool, publish_error: Optional[str]) -> None: + """Persist the audit record, the registry event log and the status sync. + + All three are best effort and all three happen whether or not the publish + succeeded — that is the contract G-1/G-2/G-4 pin down. + """ + mqtt_common.append_event(prepared.job_id, { + "event": "published", + "source_event": prepared.event, + "seq": prepared.seq, + "topic": prepared.topic, + "retain": prepared.retain, + "timestamp": prepared.payload["timestamp"], + "detail": prepared.detail, + "payload": prepared.payload, + "published": publish_ok, + "publish_error": publish_error, + }) + + registry.append_event(prepared.job_id, registry_dir, prepared.payload) + new_status = EVENT_TO_STATUS.get(prepared.event) + if new_status: + try: + mqtt_common.update_job_status(prepared.job_id, registry_dir, status=new_status) + except Exception as exc: # pragma: no cover - best effort + logger.warning("status sync failed: %s", exc) + + def _resolve_job_id(args) -> Optional[str]: if args.pick_pending: return registry.pick_pending(args.agent_session, args.registry_dir) @@ -168,26 +247,20 @@ def main(argv=None) -> int: return 1 try: - job = load_job(job_id, args.registry_dir) + prepared = prepare_event( + job_id=job_id, + event=args.event, + detail=args.detail, + data=data, + registry_dir=args.registry_dir, + retained=args.retained, + ) except FileNotFoundError as exc: logger.error("%s", exc) return 1 - config = broker_config_from_job(job) - topic = job.get("topic_prefix") - topic = f"{topic}/events" if topic else events_topic_for(job_id) - seq = next_seq(job_id, args.registry_dir) - payload = build_payload( - job_id=job_id, - seq=seq, - event=args.event, - detail=args.detail, - data=data, - auth_token=job.get("auth_token"), - ) - body = json.dumps(payload, ensure_ascii=False).encode("utf-8") - retain = args.retained or args.event in TERMINAL_EVENTS - + # Policy Note: seq consumption on failure is intentional (see prepare_event) + # and the failure is documented explicitly in the audit record. publish = with_retry( _publish_once, attempts=args.attempts, @@ -196,43 +269,19 @@ def main(argv=None) -> int: publish_ok = True publish_error: Optional[str] = None try: - publish(config, topic, body, retain) + publish(prepared.config, prepared.topic, prepared.body, prepared.retain) except Exception as exc: publish_ok = False publish_error = str(exc) logger.error("publish failed after %d attempts: %s", args.attempts, exc) - # Persistent audit log: record the exact payload we put on the wire (or intended to). - # Best-effort (isolated inside append_event) — never fails the publish. - # Policy Note: Seq consumption on failure is intentionally maintained for monotonic replay - # defense (> highest accepted seq), with failure documented explicitly in the audit record. - mqtt_common.append_event(job_id, { - "event": "published", - "source_event": args.event, - "seq": seq, - "topic": topic, - "retain": retain, - "timestamp": payload["timestamp"], - "detail": args.detail, - "payload": payload, - "published": publish_ok, - "publish_error": publish_error, - }) - - # Best-effort side effects: registry status sync + (debug) event log. Never - # fail the publish on these. - registry.append_event(job_id, args.registry_dir, payload) - new_status = EVENT_TO_STATUS.get(args.event) - if new_status: - try: - mqtt_common.update_job_status(job_id, args.registry_dir, status=new_status) - except Exception as exc: # pragma: no cover - best effort - logger.warning("status sync failed: %s", exc) + record_outcome(prepared, args.registry_dir, publish_ok, publish_error) if not publish_ok: return 2 - logger.info("published %s seq=%d job=%s retain=%s", args.event, seq, job_id, retain) + logger.info("published %s seq=%d job=%s retain=%s", + args.event, prepared.seq, job_id, prepared.retain) return 0 diff --git a/.agents/skills/lib_py/delegation/publisher.py b/.agents/skills/lib_py/delegation/publisher.py new file mode 100644 index 0000000..ece8fc5 --- /dev/null +++ b/.agents/skills/lib_py/delegation/publisher.py @@ -0,0 +1,169 @@ +"""Publish transports: one connection per event, or one connection reused. + +Phase 1 ships the persistent transport as a prototype only. Nothing in +production uses it: the CLI and the shell helpers exit after a single publish, +so reusing a connection buys them nothing. The consumer is the phase 2a engine, +which is long-running and pays the CONNACK cost on every event today. + +Both transports raise on failure so the caller's `with_retry` policy stays the +single place that decides how many attempts an event gets. +""" +from __future__ import annotations + +import logging +import time +from typing import Optional, Tuple + +from lib_py.delegation import mqtt_common + +logger = logging.getLogger("delegate_job.publisher") + +CONNECT_ACK_TIMEOUT = 10 # seconds to wait for CONNACK +PUBLISH_ACK_TIMEOUT = 5 # seconds to wait for QoS-1 PUBACK + + +class _BasePublisher(object): + """Shared retry wrapper so both transports report outcomes identically.""" + + def publish(self, prepared): # pragma: no cover - interface + raise NotImplementedError + + def publish_with_retry(self, prepared, attempts=3) -> Tuple[bool, Optional[str]]: + runner = mqtt_common.with_retry( + self.publish, + attempts=attempts, + exceptions=(OSError, TimeoutError, ConnectionError, ValueError), + ) + try: + runner(prepared) + except Exception as exc: + logger.error("publish failed after %d attempts: %s", attempts, exc) + return False, str(exc) + return True, None + + def open(self): + return self + + def close(self): + return None + + +class OneShotPublisher(_BasePublisher): + """Connect, publish, wait for the ACK, disconnect. The current behaviour.""" + + def __init__(self, client_factory=None): + self.client_factory = client_factory or mqtt_common.make_client + self.connect_count = 0 + + def publish(self, prepared): + client = self.client_factory("publisher", prepared.config) + self.connect_count += 1 + connected = {"rc": None} + + def on_connect(_c, _u, _flags, reason_code, _props): + connected["rc"] = reason_code + + client.on_connect = on_connect + client.connect(prepared.config.host, prepared.config.port, prepared.config.keepalive) + client.loop_start() + try: + deadline = time.monotonic() + CONNECT_ACK_TIMEOUT + while connected["rc"] is None and time.monotonic() < deadline: + time.sleep(0.05) + if connected["rc"] is None: + raise TimeoutError("no CONNACK from broker") + if mqtt_common.reason_code_value(connected["rc"]) != 0: + raise ConnectionError("broker refused connection: rc={}".format(connected["rc"])) + info = client.publish(prepared.topic, payload=prepared.body, + qos=1, retain=prepared.retain) + info.wait_for_publish(timeout=PUBLISH_ACK_TIMEOUT) + if not info.is_published(): + raise TimeoutError("publish not acknowledged within timeout") + finally: + client.loop_stop() + try: + client.disconnect() + except Exception: # pragma: no cover - disconnect best effort + pass + + +class PersistentPublisher(_BasePublisher): + """One connection reused across events; paho handles reconnect underneath. + + Bound to a single broker: the connection is established from the first + event's config, so an event for a job on a different broker is refused + rather than silently published to the wrong place. + """ + + def __init__(self, client_factory=None): + self.client_factory = client_factory or mqtt_common.make_client + self.connect_count = 0 + self._client = None + self._config = None + self._connected = {"rc": None} + + def __enter__(self): + return self.open() + + def __exit__(self, exc_type, exc, tb): + self.close() + return False + + def open(self): + return self + + def _ensure_client(self, config): + if self._client is not None: + if (config.host, config.port) != (self._config.host, self._config.port): + raise ValueError( + "PersistentPublisher is bound to {}:{}; refusing an event for {}:{}".format( + self._config.host, self._config.port, config.host, config.port) + ) + return self._client + + client = self.client_factory("publisher", config) + self.connect_count += 1 + + def on_connect(_c, _u, _flags, reason_code, _props): + self._connected["rc"] = reason_code + + client.on_connect = on_connect + client.connect(config.host, config.port, config.keepalive) + client.loop_start() + reconnect_delay_set = getattr(client, "reconnect_delay_set", None) + if reconnect_delay_set is not None: + reconnect_delay_set(min_delay=1, max_delay=16) + deadline = time.monotonic() + CONNECT_ACK_TIMEOUT + while self._connected["rc"] is None and time.monotonic() < deadline: + time.sleep(0.05) + if self._connected["rc"] is None: + client.loop_stop() + raise TimeoutError("no CONNACK from broker") + if mqtt_common.reason_code_value(self._connected["rc"]) != 0: + client.loop_stop() + raise ConnectionError("broker refused connection: rc={}".format(self._connected["rc"])) + self._client = client + self._config = config + return client + + def publish(self, prepared): + client = self._ensure_client(prepared.config) + info = client.publish(prepared.topic, payload=prepared.body, + qos=1, retain=prepared.retain) + info.wait_for_publish(timeout=PUBLISH_ACK_TIMEOUT) + if not info.is_published(): + # Surfaced to the caller's with_retry; paho reconnects underneath. + raise TimeoutError("publish not acknowledged within timeout") + + def close(self): + if self._client is None: + return + try: + self._client.loop_stop() + self._client.disconnect() + except Exception: # pragma: no cover - best effort + pass + finally: + self._client = None + self._config = None + self._connected = {"rc": None} diff --git a/tests/test_issue5_phase1_delegation.py b/tests/test_issue5_phase1_delegation.py index 0abdf13..2cda746 100644 --- a/tests/test_issue5_phase1_delegation.py +++ b/tests/test_issue5_phase1_delegation.py @@ -720,3 +720,163 @@ def test_publish_failure_semantics_match_across_entry_points(mam_sandbox, entry) published = [e for e in events if e.get("event") == "published"] assert published and published[-1]["published"] is False assert published[-1]["publish_error"] + + +# ============================================================================== +# 6. P1-03 — transports, parity and the benchmark harness (no network at all) +# ============================================================================== + +def _prepared(topic="python/mqtt/jobs/j/events", host="broker.example", port=1883): + from lib_py.delegation import mqtt_common, publish_event + + config = mqtt_common.broker_config_from_env({"host": host, "port": port}) + return publish_event.PreparedEvent( + job_id="j", event="progress", detail="d", seq=1, topic=topic, retain=False, + payload={"seq": 1}, body=b'{"seq": 1}', config=config, + ) + + +def _fake_factory(counters, publishes_ok=True, connect_rc=0): + from lib_py.delegation.bench import FakeClient, FakeInfo + + def factory(_role, _config): + client = FakeClient(connect_latency=0.0, counters=counters) + if not publishes_ok: + def unacked(*_a, **_k): + counters["publish"] = counters.get("publish", 0) + 1 + return FakeInfo(published=False) + client.publish = unacked + if connect_rc: + original = client.connect + + def connect(host, port, keepalive=60): + original(host, port, keepalive) + client.on_connect(client, None, None, connect_rc, None) + client.connect = connect + return client + return factory + + +def test_persistent_publisher_reuses_one_connection(): + """The whole point of the prototype: N publishes, 1 connection.""" + from lib_py.delegation.publisher import OneShotPublisher, PersistentPublisher + + prepared = _prepared() + one_counters, persist_counters = {}, {} + one = OneShotPublisher(client_factory=_fake_factory(one_counters)) + persist = PersistentPublisher(client_factory=_fake_factory(persist_counters)) + try: + for _ in range(5): + assert one.publish_with_retry(prepared, attempts=1) == (True, None) + assert persist.publish_with_retry(prepared, attempts=1) == (True, None) + finally: + persist.close() + assert one_counters["connect"] == 5 + assert persist_counters["connect"] == 1 + assert one_counters["publish"] == persist_counters["publish"] == 5 + + +def test_transport_parity_of_registry_and_audit_records(mam_sandbox, monkeypatch): + """Same input, two transports: registry and audit records must match field for field.""" + from lib_py.delegation import mqtt_common, publish_event, registry + from lib_py.delegation.publisher import OneShotPublisher, PersistentPublisher + + monkeypatch.chdir(mam_sandbox) + reg_dir = str(Path(mam_sandbox) / ".mam" / "jobs") + os.makedirs(reg_dir, exist_ok=True) + + results = {} + for label, publisher in ( + ("one_shot", OneShotPublisher(client_factory=_fake_factory({}))), + ("persistent", PersistentPublisher(client_factory=_fake_factory({}))), + ): + job_id = registry.register_job("parity " + label, registry_dir=reg_dir) + prepared = publish_event.prepare_event( + job_id=job_id, event="completed", detail="done", registry_dir=reg_dir, + ) + ok, err = publisher.publish_with_retry(prepared, attempts=1) + publish_event.record_outcome(prepared, reg_dir, ok, err) + if hasattr(publisher, "close"): + publisher.close() + record = mqtt_common.load_job(job_id, reg_dir) + events = list(mqtt_common.iter_logged_events(job_id, mqtt_common.get_logs_dir())) + audit = [e for e in events if e.get("event") == "published"][-1] + results[label] = { + "ok": ok, + "status": record["status"], + "last_seq": record.get("last_seq"), + "retain": audit["retain"], + "published": audit["published"], + "seq": audit["seq"], + "has_sig": "hmac_sig" in audit["payload"].get("data", {}), + "source_event": audit["source_event"], + } + + assert results["one_shot"] == results["persistent"], results + assert results["one_shot"]["retain"] is True, "terminal events must retain" + assert results["one_shot"]["status"] == "completed" + + +def test_persistent_publisher_unacked_publish_retries_then_records_failure(mam_sandbox, monkeypatch): + """No PUBACK -> exception -> with_retry -> record_outcome(published=False).""" + from lib_py.delegation import mqtt_common, publish_event, registry + from lib_py.delegation.publisher import PersistentPublisher + + monkeypatch.chdir(mam_sandbox) + reg_dir = str(Path(mam_sandbox) / ".mam" / "jobs") + os.makedirs(reg_dir, exist_ok=True) + job_id = registry.register_job("unacked", registry_dir=reg_dir) + prepared = publish_event.prepare_event( + job_id=job_id, event="completed", detail="d", registry_dir=reg_dir, + ) + counters = {} + publisher = PersistentPublisher(client_factory=_fake_factory(counters, publishes_ok=False)) + try: + ok, err = publisher.publish_with_retry(prepared, attempts=3) + finally: + publisher.close() + assert ok is False and "not acknowledged" in err + assert counters["publish"] == 3, "every attempt must reach the transport" + assert counters["connect"] == 1, "retries reuse the connection" + + publish_event.record_outcome(prepared, reg_dir, ok, err) + record = mqtt_common.load_job(job_id, reg_dir) + assert record["status"] == "completed", "status syncs even when the publish failed" + assert record.get("last_seq") == prepared.seq + audit = [e for e in mqtt_common.iter_logged_events(job_id, mqtt_common.get_logs_dir()) + if e.get("event") == "published"][-1] + assert audit["published"] is False and audit["publish_error"] + + +def test_persistent_publisher_refuses_a_second_broker(): + from lib_py.delegation.publisher import PersistentPublisher + + publisher = PersistentPublisher(client_factory=_fake_factory({})) + try: + assert publisher.publish_with_retry(_prepared(host="a.example"), attempts=1)[0] is True + ok, err = publisher.publish_with_retry(_prepared(host="b.example"), attempts=1) + assert ok is False + assert "bound to a.example" in err, err + finally: + publisher.close() + + +def test_bench_fake_transport_schema_and_counts(): + from lib_py.delegation import bench + + result = bench.run_benchmark(n=7, transport="fake", connect_latency_ms=1.0) + assert result["transport"] == "fake" + for arm in ("one_shot", "persistent"): + assert result[arm]["publishes"] == 7 + assert result[arm]["failures"] == 0 + assert set(result[arm]["latency_ms"]) == {"p50", "p95", "max"} + assert result["one_shot"]["client_connects"] == 7 + assert result["persistent"]["client_connects"] == 1 + assert result["persistent"]["latency_ms"]["p50"] < result["one_shot"]["latency_ms"]["p50"] + + +def test_bench_broker_transport_requires_a_job(): + from lib_py.delegation import bench + + with pytest.raises(ValueError): + bench.run_benchmark(n=1, transport="broker")