From f5c9173677f5a978b604e2ec43edc9c1c84acabb Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 5 Nov 2026 19:33:08 +0000 Subject: [PATCH] Attribute HTTP client instances and synchronize worker observations --- .../worker_startup_reconstruction/README.md | 21 +++- tests/test_worker_observation.py | 105 +++++++++++++++++ tests/worker_observation.py | 109 ++++++++++++------ 3 files changed, 193 insertions(+), 42 deletions(-) diff --git a/tests/fixtures/worker_startup_reconstruction/README.md b/tests/fixtures/worker_startup_reconstruction/README.md index ddd88ef..321c167 100644 --- a/tests/fixtures/worker_startup_reconstruction/README.md +++ b/tests/fixtures/worker_startup_reconstruction/README.md @@ -19,8 +19,9 @@ truncated or invalid, separates existing worker-history requests from submissions, and reports HTTP-buffer and milestone-event loss. These are availability/retention flags, not end-to-end phase verdicts. A complete JSON metadata record does not prove task identity, hydration, handler admission or -publication. Additional clients remain grouped, and the before/after teardown -SQL/API comparator and full phase linkage are still pending. The pinned sources +publication. HTTP clients now have separate bounded instance labels as well as +the existing role labels. Before/after teardown SQL/API collection and full +phase linkage are still pending. The pinned sources in `proposal.json` remain the historical baseline; this revision is not a new historical execution input or qualification receipt. @@ -71,10 +72,18 @@ incompatible harness is a setup finding, not permission to patch the SDK. handler, a sent request or a response without an authoritative recorded result does not prove durable publication. Match the original task/attempt/run against the Server comparator and terminal history. -- Existing primary HTTP clients have explicit labels; all other clients share - the label `additional`. Capture their actual discovery/query-poll coverage, - while retaining that limit on per-client attribution. A missing cloned-client - trace cannot count as full coverage. +- Existing primary HTTP clients have explicit role labels; other clients keep + the role `additional`. A separate weak-reference registry gives each observed + HTTP client a stable instance label and counts its routes, without retaining + the client or treating a reused integer object ID as the same instance. Its + lifetime count is capped at the observation limit. Requests beyond that cap + have no instance label and explicitly truncate client attribution. Known + clients retain their labels after overflow; no unknown request is merged into + one. Recorder updates and bounded snapshot copies share a lock across the + query and workflow threads; no network await is performed while holding it. + These are observed HTTP identities, not proof of a Client clone's parent, + worker registration or an expected-but-unobserved cloned-client trace. Missing + clone coverage still cannot count as complete end-to-end observation. The driver now captures the same bounded observation before and after shutdown on success and failure. Each capture has its own monotonic bracket; capture diff --git a/tests/test_worker_observation.py b/tests/test_worker_observation.py index 9cbe5b1..affe8f8 100644 --- a/tests/test_worker_observation.py +++ b/tests/test_worker_observation.py @@ -1,8 +1,11 @@ from __future__ import annotations import asyncio +import gc import itertools import json +import threading +import weakref from typing import Any import httpx @@ -54,6 +57,108 @@ async def handle(request: httpx.Request) -> httpx.Response: assert "SECRET" not in json.dumps(report) +@pytest.mark.asyncio +async def test_additional_clients_have_stable_distinct_bounded_instance_labels(monkeypatch: Any) -> None: + observation = WorkerObservation() + observation.install_http(monkeypatch) + transport = httpx.MockTransport(lambda request: httpx.Response(200, json={"task": None})) + async with httpx.AsyncClient(transport=transport) as first, httpx.AsyncClient(transport=transport) as second: + await first.get("https://example.test/api/cluster/info?token=SECRET") + await first.post("https://example.test/api/worker/query-tasks/poll") + await second.post("https://example.test/api/worker/query-tasks/poll") + report = observation.report() + assert [entry["client"] for entry in report["recent"]] == ["additional"] * 3 + assert [entry["client_instance"] for entry in report["recent"]] == ["http-1", "http-1", "http-2"] + assert report["client_instances"] == { + "http-1": {"counts": {"GET discovery": 1, "POST query_poll": 1}}, + "http-2": {"counts": {"POST query_poll": 1}}, + } + assert report["completeness"]["client_attribution"] == "complete" + assert "SECRET" not in json.dumps(report) + + +@pytest.mark.asyncio +async def test_client_registry_does_not_retain_clients_or_reuse_their_labels(monkeypatch: Any) -> None: + observation = WorkerObservation() + observation.install_http(monkeypatch) + async with httpx.AsyncClient(transport=httpx.MockTransport(lambda request: httpx.Response(200))) as first: + await first.get("https://example.test/api/cluster/info") + original = weakref.ref(first) + del first + gc.collect() + assert original() is None + async with httpx.AsyncClient(transport=httpx.MockTransport(lambda request: httpx.Response(200))) as second: + # Reproduce a reused integer ID without relying on allocator timing. + observation._client_instances[id(second)] = (original, "http-1") + await second.get("https://example.test/api/cluster/info") + report = observation.report() + assert [entry["client_instance"] for entry in report["recent"]] == ["http-1", "http-2"] + assert len(report["client_instances"]) == 2 + + +@pytest.mark.asyncio +async def test_client_attribution_limit_never_groups_new_clients_as_a_known_instance(monkeypatch: Any) -> None: + observation = WorkerObservation(limit=1) + observation.install_http(monkeypatch) + transport = httpx.MockTransport(lambda request: httpx.Response(200)) + async with httpx.AsyncClient(transport=transport) as first, httpx.AsyncClient(transport=transport) as second: + await first.get("https://example.test/api/cluster/info") + await second.get("https://example.test/api/cluster/info") + overflow = observation.report() + assert overflow["recent"][0]["client_instance"] is None + assert overflow["recent"][0]["client_instance_state"] == "truncated" + await first.get("https://example.test/api/cluster/info") + report = observation.report() + assert report["recent"][0]["client_instance"] == "http-1" + assert report["recent"][0]["client_instance_state"] == "complete" + assert report["completeness"]["client_attribution"] == "truncated" + assert len(report["client_instances"]) == 1 and report["unattributed_client_requests"] == 1 + assert overflow["client_instances"]["http-1"]["counts"]["GET discovery"] == 1 + + +def test_query_thread_and_workflow_thread_share_consistent_snapshots(monkeypatch: Any) -> None: + observation = WorkerObservation() + observation.install_http(monkeypatch) + entered = [threading.Event(), threading.Event()] + release = threading.Event() + errors: list[BaseException] = [] + + def run(index: int, family: str) -> None: + async def execute(): + def reply(request: httpx.Request) -> httpx.Response: + entered[index].set() + assert release.wait(5) + return httpx.Response(200, json={"task": None}) + async with httpx.AsyncClient(transport=httpx.MockTransport(reply)) as client: + await client.post(f"https://example.test/api/worker/{family}-tasks/poll") + observation.note("request_returned") + try: + asyncio.run(execute()) + except BaseException as error: + errors.append(error) + + threads = [threading.Thread(target=run, args=(index, family)) + for index, family in enumerate(("workflow", "query"))] + try: + for thread in threads: + thread.start() + assert all(event.wait(5) for event in entered) + before = observation.report() + assert before["active_count"] == len(before["active"]) == 2 + assert len({entry["client_instance"] for entry in before["active"]}) == 2 + finally: + release.set() + for thread in threads: + thread.join(5) + assert not errors and not any(thread.is_alive() for thread in threads) + after = observation.report() + assert after["requests"] == 2 and after["active_count"] == 0 + assert after["counts"] == {"POST workflow_poll": 1, "POST query_poll": 1} + assert after["milestones"]["request_returned"]["count"] == 2 + assert {entry["request"] for entry in after["recent"]} == {1, 2} + assert before["active_count"] == 2 and not before["milestones"] + + @pytest.mark.asyncio async def test_observer_follows_the_actual_client_registration_route(monkeypatch: Any) -> None: observation = WorkerObservation() diff --git a/tests/worker_observation.py b/tests/worker_observation.py index a8907c4..90983f4 100644 --- a/tests/worker_observation.py +++ b/tests/worker_observation.py @@ -4,7 +4,9 @@ import asyncio import json +import threading import time +import weakref from collections import Counter, deque from collections.abc import Callable, Coroutine from copy import deepcopy @@ -80,17 +82,37 @@ def __init__(self, *, clock: Callable[[], float] = time.monotonic, limit: int = self.evicted_active = 0 self.omitted_milestone_events = 0 self.response_metadata_states: Counter[str] = Counter() + self._lock = threading.Lock() + self._client_instances: dict[int, tuple[weakref.ReferenceType[httpx.AsyncClient], str]] = {} + self.client_instances: dict[str, dict[str, Any]] = {} + self.unattributed_client_requests = 0 def note(self, phase: str, task: Any = None) -> None: - if phase not in self.milestones and len(self.milestones) >= self.limit: - self.omitted_milestone_events += 1 - return - entry = {"at": self.clock() - self.started, **task_identity(task)} - milestone = self.milestones.setdefault(phase, {"count": 0, "first": entry}) - if milestone["count"] >= 2: - self.omitted_milestone_events += 1 - milestone["count"] += 1 - milestone["last"] = entry + with self._lock: + if phase not in self.milestones and len(self.milestones) >= self.limit: + self.omitted_milestone_events += 1 + return + entry = {"at": self.clock() - self.started, **task_identity(task)} + milestone = self.milestones.setdefault(phase, {"count": 0, "first": entry}) + if milestone["count"] >= 2: + self.omitted_milestone_events += 1 + milestone["count"] += 1 + milestone["last"] = entry + + def _client_instance(self, client: httpx.AsyncClient, key: str) -> str | None: + """Called under the lock; weak references never keep an HTTP client alive.""" + previous = self._client_instances.get(id(client)) + if previous is not None and previous[0]() is client: + label = previous[1] + elif len(self.client_instances) < self.limit: + label = f"http-{len(self.client_instances) + 1}" + self._client_instances[id(client)] = (weakref.ref(client), label) + self.client_instances[label] = {"counts": Counter()} + else: + self.unattributed_client_requests += 1 + return None + self.client_instances[label]["counts"][key] += 1 + return label def install_http(self, monkeypatch: Any) -> None: original_send = httpx.AsyncClient.send @@ -98,43 +120,51 @@ def install_http(self, monkeypatch: Any) -> None: async def observed_send( client: httpx.AsyncClient, request: httpx.Request, *args: Any, **kwargs: Any, ) -> httpx.Response: - self.total += 1 - request_id = self.total started = self.clock() route = route_name(request) method = request.method if request.method in {"GET", "POST", "PUT", "PATCH", "DELETE"} else "OTHER" - self.counts[f"{method} {route}"] += 1 - entry: dict[str, Any] = { - "request": request_id, "client": self.clients.get(id(client), "additional"), - "method": method, "route": route, "started": started - self.started, - "response_metadata_state": "missing", - } - if len(self.active) == self.limit: - self.active.pop(next(iter(self.active))) - self.evicted_active += 1 - self.active[request_id] = entry - self.active_count += 1 + key = f"{method} {route}" + with self._lock: + self.total += 1 + request_id = self.total + self.counts[key] += 1 + instance = self._client_instance(client, key) + entry: dict[str, Any] = { + "request": request_id, "client": self.clients.get(id(client), "additional"), + "client_instance": instance, + "client_instance_state": "complete" if instance is not None else "truncated", + "method": method, "route": route, "started": started - self.started, + "response_metadata_state": "missing", + } + if len(self.active) == self.limit: + self.active.pop(next(iter(self.active))) + self.evicted_active += 1 + self.active[request_id] = entry + self.active_count += 1 try: response = await original_send(client, request, *args, **kwargs) - entry["status"] = response.status_code - if 200 <= response.status_code < 300: - self.successful[f"{method} {route}"] += 1 + metadata: dict[str, Any] = {"status": response.status_code} if response.is_stream_consumed: - entry["response_metadata_state"] = self._response_metadata(entry, response) + metadata["response_metadata_state"] = self._response_metadata(metadata, response) + with self._lock: + entry.update(metadata) + if 200 <= response.status_code < 300: + self.successful[key] += 1 return response except BaseException as error: - entry["exception"] = type(error).__name__ + with self._lock: + entry["exception"] = type(error).__name__ raise finally: - entry["elapsed"] = self.clock() - started - key = f"{method} {route}" - self.max_elapsed[key] = max(self.max_elapsed.get(key, 0), entry["elapsed"]) - self.active_count -= 1 - self.active.pop(request_id, None) - self.response_metadata_states[entry["response_metadata_state"]] += 1 - if len(self.recent) == self.limit: - self.omitted_completed += 1 - self.recent.append(entry) + with self._lock: + entry["elapsed"] = self.clock() - started + self.max_elapsed[key] = max(self.max_elapsed.get(key, 0), entry["elapsed"]) + self.active_count -= 1 + self.active.pop(request_id, None) + self.response_metadata_states[entry["response_metadata_state"]] += 1 + if len(self.recent) == self.limit: + self.omitted_completed += 1 + self.recent.append(entry) monkeypatch.setattr(httpx.AsyncClient, "send", observed_send) @@ -176,6 +206,10 @@ def _response_metadata(entry: dict[str, Any], response: httpx.Response) -> str: return state def report(self, *, include_recent: bool = True) -> dict[str, Any]: + with self._lock: + return self._report(include_recent=include_recent) + + def _report(self, *, include_recent: bool) -> dict[str, Any]: return deepcopy({ "elapsed": self.clock() - self.started, "requests": self.total, "counts": dict(self.counts), "milestones": self.milestones, @@ -190,11 +224,14 @@ def report(self, *, include_recent: bool = True) -> dict[str, Any]: ), "active_http": "truncated" if self.evicted_active else "complete", "milestones": "truncated" if self.omitted_milestone_events else "complete", + "client_attribution": "truncated" if self.unattributed_client_requests else "complete", }, "omitted_completed": self.omitted_completed, "evicted_active": self.evicted_active, "omitted_milestone_events": self.omitted_milestone_events, "response_metadata_states": dict(self.response_metadata_states), + "client_instances": self.client_instances, + "unattributed_client_requests": self.unattributed_client_requests, })