From 621b0603ee651946fa9ba188c0c852989f61130c Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Fri, 9 Oct 2026 21:27:49 -0400 Subject: [PATCH 1/3] Say what the collector refused, fit requests to its limit, register a lost run again Three follow-ups to the settled-rejection rule in the telemetry service's collector delivery, made in step with Microcosm's copy of this code. A refused registration printed the "credential was not accepted as a PolicyEngine organization member" warning, because `_make_local_only` defaulted to it. That was false for a 409 (the owner registering with other metadata), a 422 and any other settled 4xx, and for a 403 on a registration or on events, where the collector had accepted the login and refused it the run. A rejected registration now names its HTTP status, a 403 names the Hugging Face user, and the message no longer has a default. A batch was bounded by count only. A full batch of the largest events (non-ASCII messages, details at their limit) is over the collector's 1 MiB request limit, which it answers with 413 for the whole request, and that made the run local-only. A batch is now the longest run of leading events whose request body fits, measured with the serialization `_http_post` sends. An event too large to send alone is sent alone, for the collector to refuse. A 404 for a run's events means the collector does not have a run this service registered. The service now registers the run again and sends the same batch on the next pass; a second 404 with no batch acknowledged in between makes the run local-only, as before. Stacked on #6 (0038e99). Co-Authored-By: Claude Opus 5.5 --- README.md | 4 +- ...etry-collector-delivery-followups.fixed.md | 1 + .../telemetry/service/collector.py | 64 +- .../telemetry/service/constants.py | 13 + tests/telemetry/test_collector_delivery.py | 920 ++++++++++++++++++ tests/telemetry/test_delivery.py | 18 +- tests/telemetry/test_foreign_runs.py | 164 +++- tests/test_collector_contract.py | 315 +++++- 8 files changed, 1434 insertions(+), 65 deletions(-) create mode 100644 changelog.d/telemetry-collector-delivery-followups.fixed.md create mode 100644 tests/telemetry/test_collector_delivery.py diff --git a/README.md b/README.md index b1ed40c..a2b3f74 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,7 @@ The service interprets requests such as a stage change or build failure, timesta The adapter bounds socket calls and reports telemetry failures without failing the build. It converts Python arguments into JSON but does not process their telemetry meaning. Network requests run on the service worker, outside the message-handling lock. -The service uses the existing Hugging Face credential from the environment or local login cache. The collector validates PolicyEngine membership before issuing a short-lived write credential, and the first Hugging Face user to register a run owns it. Missing or rejected credentials permanently mark the service's own producer's events as local-only. Transient collector failures (a timeout, a rate limit, a server error) retain events for retry. A refusal of a run's registration or events that is about the run and not the login (a 4xx other than 401, 403, 408 and 429) makes that run local-only instead, whichever service sends it. The production destination belongs to the package; development overrides accept loopback addresses only. +The service uses the existing Hugging Face credential from the environment or local login cache. The collector validates PolicyEngine membership before issuing a short-lived write credential, and the first Hugging Face user to register a run owns it. Missing or rejected credentials permanently mark the service's own producer's events as local-only. Transient collector failures (a timeout, a rate limit, a server error) retain events for retry. A refusal of a run's registration or events that is about the run and not the login (a 4xx other than 401, 403, 408 and 429) makes that run local-only instead, whichever service sends it. The warning a service prints for its own run says which request the collector refused and with what status; only a refused token exchange is reported as a credential the collector did not accept. One answer is tried again first: when the collector answers 404 for a run's events, it does not have a run the service registered (after a database restore, say), so the service registers the run again and sends the same events. A second 404 with no events acknowledged in between makes the run local-only. Events are sent in order, at most 100 to a request and never more than the collector's 1 MiB request limit, so a queue of large events goes in several requests instead of one the collector refuses. The production destination belongs to the package; development overrides accept loopback addresses only. Builds that supply the same queue path share one queue, so it also holds other builds' runs. A service's login decides only its own run. Each service holds an advisory lock file beside the queue, in `.leases/`, for its whole life. Another service delivers that run's remaining events only once the lock is free, which means the producing service has exited. It does so with its own login, and only while the collector accepts that login for the run; otherwise the run stays pending for another service. A run with no lock file, from a service that predates them, is left alone until it has been idle for 15 minutes. Microcosm's telemetry service keeps the same lock files from PolicyEngine/microcosm#1177 on, so services from both packages recognise each other's runs. The shutdown drain waits only for the service's own run. @@ -35,7 +35,7 @@ bash tools/check-artifacts.sh Tests isolate ambient credentials and use synthetic identities. The artifact check builds one source archive and wheel, installs that wheel into a clean environment away from the source checkout, verifies that the host does not import telemetry or its dependencies, and exercises the real telemetry service process and packaged migrations. -The collector compatibility job tests the installed wheel against a pinned dashboard collector and disposable PostgreSQL. It covers the HTTP exchange, event ingestion, dashboard run documents, rejected identities, and duplicate delivery after a lost acknowledgement. Local execution requires that pinned collector checkout and a loopback test database; `tools/check-collector.sh` verifies these inputs. Never point this test at a deployed database. +The collector compatibility job tests the installed wheel against a pinned dashboard collector and disposable PostgreSQL. It covers the HTTP exchange, event ingestion, dashboard run documents, rejected identities, duplicate delivery after a lost acknowledgement, refused registrations and events, a run the collector does not have, and requests at the collector's size limit. Local execution requires that pinned collector checkout and a loopback test database; `tools/check-collector.sh` verifies these inputs. Never point this test at a deployed database. ## Publication diff --git a/changelog.d/telemetry-collector-delivery-followups.fixed.md b/changelog.d/telemetry-collector-delivery-followups.fixed.md new file mode 100644 index 0000000..e3784b7 --- /dev/null +++ b/changelog.d/telemetry-collector-delivery-followups.fixed.md @@ -0,0 +1 @@ +The telemetry service handles three collector answers better. A run whose registration the collector refuses now gets a warning that says so and names the HTTP status. Before, the warning said the Hugging Face credential was not accepted as a PolicyEngine organization member, which was false for a 409 (the run's owner registering it with other metadata), a 422 (a registration outside the collector's schema) and a 403 (another user registered the run first, on a registration or on events); the 403 warning now names the Hugging Face user. A queue of events is cut to the collector's 1 MiB request limit before it is sent. A full request of 100 of the largest events, with non-ASCII messages and details at their limit, was over it; the collector refuses such a request whole, which made the run local-only. Events still go in sequence order with none skipped, and an event too large to send alone is sent alone and left for the collector to refuse. When the collector answers 404 for a run's events, it does not have a run the service registered (after a database restore, say). The service now registers the run again and sends the same events; before, the run went local-only at once. A second 404 with no events acknowledged in between still makes it local-only, so a collector that never finds the run is asked twice and then left alone. Microcosm's telemetry service makes the same three changes, so both packages' services treat a shared queue's runs alike. diff --git a/src/microcosm_emitter/telemetry/service/collector.py b/src/microcosm_emitter/telemetry/service/collector.py index 2c333f2..a034c03 100644 --- a/src/microcosm_emitter/telemetry/service/collector.py +++ b/src/microcosm_emitter/telemetry/service/collector.py @@ -34,6 +34,7 @@ LOCAL_ONLY_REJECTED_EVENTS, LOCAL_ONLY_REJECTED_REGISTRATION, LOOPBACK_HOSTS, + MAX_EVENTS_REQUEST_BYTES, MAX_HTTP_RESPONSE_BYTES, MAX_RETRY_SECONDS, MINIMUM_TOKEN_LIFETIME_SECONDS, @@ -43,6 +44,8 @@ PRODUCTION_COLLECTOR_URL, REJECTED_CREDENTIAL_MESSAGE, REJECTED_EVENTS_MESSAGE, + REJECTED_REGISTRATION_MESSAGE, + REJECTED_RUN_USER_MESSAGE, RUN_EVENTS_PATH_TEMPLATE, RUN_REGISTRATION_PATH, TOKEN_EXCHANGE_PATH, @@ -97,6 +100,33 @@ def _decode_response(body: bytes) -> dict[str, Any]: return response if isinstance(response, dict) else {} +def _request_body(payload: Mapping[str, Any]) -> bytes: + """Serialize a request body: the bytes that are sent and that are measured.""" + + return json.dumps(payload, separators=(",", ":")).encode() + + +#: An events request holding no events: what every batch is wrapped in. +_EVENTS_ENVELOPE_BYTES = len(_request_body({"events": []})) + + +def _events_fitting_request(events: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Return the longest run of leading events whose request body fits. + + The collector refuses a larger body whole, which would settle the run as + rejected. The first event is always kept: a request is never empty, and an + event too large to send alone is the collector's to refuse. + """ + + size = _EVENTS_ENVELOPE_BYTES + for index, event in enumerate(events): + # After the first, each event also costs the comma before it. + size += len(_request_body(event)) + (1 if index else 0) + if index and size > MAX_EVENTS_REQUEST_BYTES: + return events[:index] + return events + + def _http_post( url: str, payload: Mapping[str, Any], @@ -106,7 +136,7 @@ def _http_post( ) -> tuple[int, dict[str, Any]]: request = urllib.request.Request( url, - data=json.dumps(payload, separators=(",", ":")).encode(), + data=_request_body(payload), headers={ "Authorization": f"Bearer {bearer_token}", "Content-Type": "application/json", @@ -197,6 +227,9 @@ def __init__( self._own_lease = self._hold_own_lease() self._session_token: tuple[str, float] | None = None self._registered: set[tuple[str, str]] = set() + # Runs being registered again because the collector answered that it + # does not have them; an acknowledged batch clears the run. + self._registering_again: set[tuple[str, str]] = set() self._warned_no_token = False self._warned_denied: set[tuple[str, str]] = set() self._next_attempt_at = 0.0 @@ -346,7 +379,7 @@ def _deliver(self, registration: Mapping[str, Any]) -> bool: if (run_id, producer_id) not in self._registered: if not self._register(registration, token): return False - events = self.spool.batch(run_id, producer_id) + events = _events_fitting_request(self.spool.batch(run_id, producer_id)) if not events: return False try: @@ -361,6 +394,7 @@ def _deliver(self, registration: Mapping[str, Any]) -> bool: if status == HTTPStatus.ACCEPTED: self.spool.acknowledge([event["event_id"] for event in events]) self._retry_seconds = INITIAL_RETRY_SECONDS + self._registering_again.discard(_run(registration)) return True if status == HTTPStatus.UNAUTHORIZED: self._session_token = None @@ -371,7 +405,19 @@ def _deliver(self, registration: Mapping[str, Any]) -> bool: registration, LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION, caller_dependent=True, + message=REJECTED_RUN_USER_MESSAGE.format(status=int(status)), ) + elif ( + status == HTTPStatus.NOT_FOUND + and _run(registration) not in self._registering_again + ): + # The collector no longer has a run it registered for this service + # (its database was restored, say). Registering is repeatable for + # the run's owner, so the next pass registers the run and sends the + # batch again. Not finding the run once more, with no batch + # acknowledged in between, is its settled answer. + self._registered.discard(_run(registration)) + self._registering_again.add(_run(registration)) elif _permanent_client_error(status): self._make_local_only( registration, @@ -397,6 +443,8 @@ def _collector_token(self, registration: Mapping[str, Any]) -> str | None: registration, LOCAL_ONLY_MISSING_CREDENTIAL, caller_dependent=True, + # Said once for the service above, not per run. + message=None, ) return None credential = _credential_fingerprint(hf_token) @@ -408,6 +456,7 @@ def _collector_token(self, registration: Mapping[str, Any]) -> str | None: registration, LOCAL_ONLY_REJECTED_CREDENTIAL, caller_dependent=True, + message=REJECTED_CREDENTIAL_MESSAGE, ) return None try: @@ -433,6 +482,7 @@ def _collector_token(self, registration: Mapping[str, Any]) -> str | None: registration, LOCAL_ONLY_REJECTED_CREDENTIAL, caller_dependent=True, + message=REJECTED_CREDENTIAL_MESSAGE, ) else: self._defer_retry() @@ -460,6 +510,7 @@ def _register(self, registration: Mapping[str, Any], token: str) -> bool: registration, LOCAL_ONLY_REJECTED_REGISTRATION, caller_dependent=True, + message=REJECTED_RUN_USER_MESSAGE.format(status=int(status)), ) elif status == HTTPStatus.CONFLICT or _permanent_client_error(status): # A conflict reaches only the run's owner, so it is about the @@ -468,6 +519,7 @@ def _register(self, registration: Mapping[str, Any], token: str) -> bool: registration, LOCAL_ONLY_REJECTED_REGISTRATION, caller_dependent=False, + message=REJECTED_REGISTRATION_MESSAGE.format(status=int(status)), ) else: self._defer_retry() @@ -479,7 +531,7 @@ def _make_local_only( reason: str, *, caller_dependent: bool, - message: str = REJECTED_CREDENTIAL_MESSAGE, + message: str | None, ) -> None: """Stop uploading a run, unless the reason is only this service's. @@ -488,6 +540,10 @@ def _make_local_only( service's own run, whose credential it is. Another producer's run stays pending for a service whose credential the collector accepts for it, and is not offered again with this credential. + + ``message`` is the warning for the service's own run, and has no + default: each answer says what the collector refused. ``None`` prints + nothing. """ run = _run(registration) @@ -499,7 +555,7 @@ def _make_local_only( self.spool.make_local_only(*run, reason) # The warnings say "this run": another build's run is not this build's # to report, and its reason is kept in the spool. - if own and reason != LOCAL_ONLY_MISSING_CREDENTIAL: + if own and message is not None: self._warn_denied(run, message) def _defer_retry(self) -> None: diff --git a/src/microcosm_emitter/telemetry/service/constants.py b/src/microcosm_emitter/telemetry/service/constants.py index 3c447de..c421d7e 100644 --- a/src/microcosm_emitter/telemetry/service/constants.py +++ b/src/microcosm_emitter/telemetry/service/constants.py @@ -17,6 +17,9 @@ RUN_EVENTS_PATH_TEMPLATE: Final = "/v1/runs/{run_id}/events" HTTP_USER_AGENT: Final = "microcosm-telemetry-emitter/1" MAX_HTTP_RESPONSE_BYTES: Final = 1_048_576 +#: The collector answers a request body over 1 MiB with 413, whatever is in +#: it, so a batch of events is cut to fit before it is sent. +MAX_EVENTS_REQUEST_BYTES: Final = 1_048_576 HTTP_TIMEOUT_SECONDS: Final = 5.0 UPLOAD_STATE_PENDING: Final = "pending" @@ -42,6 +45,16 @@ "events (HTTP {status}), so they will not be retried. The dataset build will " "continue." ) +REJECTED_REGISTRATION_MESSAGE: Final = ( + "Microcosm telemetry is local-only for this run: the collector rejected its " + "registration (HTTP {status}), so it will not be retried. The dataset build " + "will continue." +) +REJECTED_RUN_USER_MESSAGE: Final = ( + "Microcosm telemetry is local-only for this run: the collector refused the " + "ambient Hugging Face user for it (HTTP {status}). A run belongs to the user " + "who registered it first. The dataset build will continue." +) OWN_LEASE_UNAVAILABLE_MESSAGE: Final = ( "Microcosm telemetry is off for this run: its service could not take the " "lock that marks the run as live, and without it another build's service " diff --git a/tests/telemetry/test_collector_delivery.py b/tests/telemetry/test_collector_delivery.py new file mode 100644 index 0000000..b3a318b --- /dev/null +++ b/tests/telemetry/test_collector_delivery.py @@ -0,0 +1,920 @@ +"""What a telemetry service sends the collector, and how it takes three answers. + +The fake collector here answers as the hosted one does +(PolicyEngine/calibration-diagnostics, ``telemetry-service``): a request body +over 1 MiB gets 413 before anything in it is read; events for a run it does not +have get 404; registering a run is repeatable for the Hugging Face user who owns +it; another user gets 403, and the owner gets 409 for other metadata. + +Invariants, each checked below for a service delivering its own run: + +- A warning says what the collector refused. A refused registration names its + HTTP status, a 403 names the Hugging Face user and not organization + membership, and only a refused token exchange says the credential was not + accepted. +- An events request is never empty and never over 1 MiB unless its only event + is. Requests carry a run's events in sequence order with none skipped or + repeated, and each is the longest run of the next queued events that fits. +- A run the collector answers 404 for is registered again once; a second 404 + with no batch acknowledged in between makes the run local-only. So between + two registrations of a run there is always a 404, and from the second + registration on, an acknowledged batch as well. +""" + +from __future__ import annotations + +import itertools +import json +import threading +from collections.abc import Iterator, Mapping +from contextlib import contextmanager +from dataclasses import dataclass, field +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from types import SimpleNamespace +from typing import Any +from unittest import mock +from urllib.parse import urlsplit + +import pytest +from hypothesis import HealthCheck, example, given, settings +from hypothesis import strategies as st +from sqlalchemy.orm import Session + +from microcosm_emitter.telemetry.client import TelemetryRun +from microcosm_emitter.telemetry.protocol import ( + MAX_TELEMETRY_DETAILS_BYTES, + MAX_TELEMETRY_MESSAGE_CHARS, +) +from microcosm_emitter.telemetry.service import collector as collector_module +from microcosm_emitter.telemetry.service.collector import CollectorDelivery +from microcosm_emitter.telemetry.service.constants import ( + BATCH_SIZE, + LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION, + LOCAL_ONLY_REJECTED_CREDENTIAL, + LOCAL_ONLY_REJECTED_EVENTS, + LOCAL_ONLY_REJECTED_REGISTRATION, + MAX_EVENTS_REQUEST_BYTES, + MAX_RETRY_SECONDS, + RUN_REGISTRATION_PATH, + TOKEN_EXCHANGE_PATH, +) +from microcosm_emitter.telemetry.service.database import create_spool_engine +from microcosm_emitter.telemetry.service.models import TelemetryRunRecord +from microcosm_emitter.telemetry.service.sanitization import ( + sanitize_details, + sanitize_text, +) +from microcosm_emitter.telemetry.service.spool import EventSpool + +DEVELOPMENT_COLLECTOR = "http://127.0.0.1:8080" +#: The hosted collector's body limit, written out here and not imported: its +#: ``RequestBodyLimitMiddleware(max_bytes=1_048_576)`` refuses anything larger. +COLLECTOR_BODY_LIMIT = 1_048_576 +#: The longest identifier the collector's schema accepts. +LONGEST_IDENTIFIER = "r" * 192 +RESOURCES = { + "cpu_user_seconds": 123456.789012, + "cpu_system_seconds": 123456.789012, + "rss_bytes": 2**40, + "peak_rss_bytes": 2**40, +} +SMALL_EVENT = { + "timestamp": "2026-10-09T00:00:00+00:00", + "event_type": "stage", + "stage_id": "compile", + "status": "started", + "message": None, + "details": {}, +} + + +def _registration(run_id: str = "run-a", producer_id: str = "producer-a") -> dict: + return TelemetryRun( + run_id=run_id, + country_code="US", + pipeline="collector-delivery", + producer_id=producer_id, + ).as_registration() + + +class FakeCollector: + """The hosted collector's answers to one member's requests.""" + + def __init__(self) -> None: + self.runs: set[str] = set() + self.body_limit = COLLECTOR_BODY_LIMIT + # Whether a registration it answers 201 is kept. Off: a collector that + # accepts a run and still does not have it. + self.keeps_runs = True + # A status to answer a kind of request with, instead of its own. + self.refuse: dict[str, int] = {} + self.accepted: list[dict[str, Any]] = [] + # ("exchange" | "register" | "events", status), in the order answered. + self.answers: list[tuple[str, int]] = [] + # (body bytes, sequences) of every events request, whatever the answer. + self.event_requests: list[tuple[int, list[int]]] = [] + + def post(self, url, payload, token, *, timeout=5.0) -> tuple[int, dict]: + path = urlsplit(url).path + if path == TOKEN_EXCHANGE_PATH: + if "exchange" in self.refuse: + return self._answer("exchange", self.refuse["exchange"]) + body = {"access_token": "collector-token", "expires_in": 3600} + return self._answer("exchange", 200, body) + kind = "register" if path == RUN_REGISTRATION_PATH else "events" + size = len(collector_module._request_body(payload)) + if kind == "events": + sequences = [event["sequence"] for event in payload["events"]] + self.event_requests.append((size, sequences)) + if size > self.body_limit: + return self._answer(kind, 413) + if kind in self.refuse: + return self._answer(kind, self.refuse[kind]) + if kind == "register": + if self.keeps_runs: + self.runs.add(payload["run_id"]) + return self._answer(kind, 201, {"registered": True}) + if path.split("/")[3] not in self.runs: + return self._answer(kind, 404) + self.accepted.extend(payload["events"]) + return self._answer(kind, 202, {"accepted": len(payload["events"])}) + + def _answer(self, kind: str, status: int, body: dict | None = None): + self.answers.append((kind, status)) + return status, body if body is not None else {"detail": "refused"} + + def asked(self, kind: str) -> list[int]: + return [status for what, status in self.answers if what == kind] + + +@dataclass +class Service: + """One build's delivery, its spool, and a clock the test moves.""" + + spool: EventSpool + registration: dict[str, Any] + delivery: CollectorDelivery + clock: SimpleNamespace + queued: list[dict[str, Any]] = field(default_factory=list) + + def emit(self, event: Mapping[str, Any] = SMALL_EVENT, count: int = 1) -> None: + for _ in range(count): + self.queued.append( + self.spool.append(self.registration, event, resources=RESOURCES) + ) + + def flush(self) -> bool: + return self.delivery.flush_once() + + def flush_until_idle(self, limit: int = 50) -> None: + for _ in range(limit): + if not self.flush(): + return + raise AssertionError("delivery never went idle") + + def wait_out_any_retry_delay(self) -> None: + self.clock.now += MAX_RETRY_SECONDS + 1 + + def local_only_reason(self) -> str | None: + engine = create_spool_engine(self.spool.path) + try: + with Session(engine) as session: + run = session.get( + TelemetryRunRecord, + (self.registration["run_id"], self.registration["producer_id"]), + ) + return run.local_only_reason + finally: + engine.dispose() + + def pending(self) -> bool: + return self.spool.pending_runs() == [self.registration] + + +_RUN_NUMBERS = itertools.count() + + +def _new_registration(longest: bool = False) -> dict[str, Any]: + """A run no earlier service in this module used, so one spool serves many.""" + + number = f"{next(_RUN_NUMBERS):08d}" + if longest: + return _registration( + number + LONGEST_IDENTIFIER[len(number) :], + number + "p" * (len(LONGEST_IDENTIFIER) - len(number)), + ) + return _registration(f"run-{number}", f"producer-{number}") + + +@contextmanager +def _service( + spool: EventSpool, + collector: FakeCollector, + registration: dict[str, Any] | None = None, +) -> Iterator[Service]: + """Run one service on ``spool``, leaving none of its events queued after. + + A test that runs many services in turn gives each a new run on one spool; + a run left with events would be an orphan for the next service to adopt. + """ + + registration = registration or _new_registration() + spool.register(registration) + clock = SimpleNamespace(now=0.0) + with ( + mock.patch.object(collector_module, "_huggingface_token", lambda: "hf-member"), + mock.patch.object(collector_module, "_http_post", collector.post), + mock.patch.object( + collector_module, "time", SimpleNamespace(monotonic=lambda: clock.now) + ), + ): + delivery = CollectorDelivery( + spool, registration, development_collector_url=DEVELOPMENT_COLLECTOR + ) + service = Service(spool, registration, delivery, clock) + try: + yield service + finally: + delivery.close() + spool.acknowledge([event["event_id"] for event in service.queued]) + + +@pytest.fixture +def collector() -> FakeCollector: + return FakeCollector() + + +@pytest.fixture +def spool(tmp_path) -> Iterator[EventSpool]: + spool = EventSpool(tmp_path / "events.sqlite3") + yield spool + spool._engine.dispose() + + +@pytest.fixture +def service(spool, collector) -> Iterator[Service]: + with _service(spool, collector) as running: + yield running + + +# --- What a warning says ----------------------------------------------------- + + +@pytest.mark.parametrize( + ("status", "warning"), + [ + pytest.param( + 409, + "Microcosm telemetry is local-only for this run: the collector rejected " + "its registration (HTTP 409), so it will not be retried. The dataset " + "build will continue.\n", + id="409", + ), + pytest.param( + 422, + "Microcosm telemetry is local-only for this run: the collector rejected " + "its registration (HTTP 422), so it will not be retried. The dataset " + "build will continue.\n", + id="422", + ), + pytest.param( + 403, + "Microcosm telemetry is local-only for this run: the collector refused " + "the ambient Hugging Face user for it (HTTP 403). A run belongs to the " + "user who registered it first. The dataset build will continue.\n", + id="403", + ), + ], +) +def test_a_rejected_registration_warning_says_what_the_collector_refused( + service, collector, capsys, status, warning +) -> None: + """The collector answers 409 to the run's owner for other metadata, 422 for + a registration outside its schema, and 403 when another Hugging Face user + registered the run first. None of them is about organization membership, + which the token exchange had already accepted.""" + collector.refuse["register"] = status + service.emit() + + assert not service.flush() + service.wait_out_any_retry_delay() + assert not service.flush() + + assert service.local_only_reason() == LOCAL_ONLY_REJECTED_REGISTRATION + assert collector.asked("register") == [status] + assert capsys.readouterr().err == warning + + +def _expected_warning(request: str, status: int) -> str | None: + if request == "exchange": + return ( + "Microcosm telemetry is local-only for this run: the ambient Hugging " + "Face credential was not accepted as a PolicyEngine organization " + "member. The dataset build will continue.\n" + ) + if status == 403: + return ( + "Microcosm telemetry is local-only for this run: the collector refused " + "the ambient Hugging Face user for it (HTTP 403). A run belongs to the " + "user who registered it first. The dataset build will continue.\n" + ) + what, pronoun = ( + ("registration", "it") if request == "register" else ("events", "they") + ) + return ( + "Microcosm telemetry is local-only for this run: the collector rejected " + f"its {what} (HTTP {status}), so {pronoun} will not be retried. The " + "dataset build will continue.\n" + ) + + +@pytest.mark.parametrize("request_kind", ["exchange", "register", "events"]) +def test_every_refusal_prints_one_warning_about_the_request_refused( + spool, capsys, request_kind +) -> None: + """Over every 4xx that makes the service's own run local-only, on each of + the three requests: exactly one warning, worded for that request. A refused + token exchange is the only answer about the credential.""" + reasons = { + "exchange": LOCAL_ONLY_REJECTED_CREDENTIAL, + "register": LOCAL_ONLY_REJECTED_REGISTRATION, + "events": LOCAL_ONLY_REJECTED_EVENTS, + } + statuses = ( + (401, 403) + if request_kind == "exchange" + else [s for s in range(400, 500) if s not in (401, 408, 429)] + ) + for status in statuses: + collector = FakeCollector() + collector.refuse[request_kind] = status + with _service(spool, collector) as service: + service.emit() + for _ in range(3): + service.flush() + service.wait_out_any_retry_delay() + + expected_reason = reasons[request_kind] + if request_kind == "events" and status == 403: + expected_reason = LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION + assert service.local_only_reason() == expected_reason, status + warning = capsys.readouterr().err + assert warning == _expected_warning(request_kind, status), status + if request_kind != "exchange": + assert "credential" not in warning, status + assert "organization member" not in warning, status + + +# --- How large a request is -------------------------------------------------- + + +def _json_bytes(value: Any) -> int: + return len(json.dumps(value, separators=(",", ":")).encode()) + + +def _details(keys: int, character: str, length: int) -> dict[str, Any]: + """Details bounded as every event's are before it is queued.""" + + return sanitize_details({f"k{index}": character * length for index in range(keys)}) + + +def _largest_details() -> dict[str, Any]: + """Details of exactly ``MAX_TELEMETRY_DETAILS_BYTES``, the most kept whole.""" + + details: dict[str, Any] = {} + while True: + grown = {**details, f"k{len(details)}": "x" * 190} + if _json_bytes(grown) > MAX_TELEMETRY_DETAILS_BYTES - 16: + break + details = grown + # One more key takes the object to the limit exactly: ,"pad":"…" + padding = MAX_TELEMETRY_DETAILS_BYTES - _json_bytes(details) - len(',"pad":""') + details["pad"] = "x" * padding + assert _json_bytes(details) == MAX_TELEMETRY_DETAILS_BYTES + assert sanitize_details(details) == details + return details + + +def _bounded_event(message: str | None, details: Mapping[str, Any]) -> dict[str, Any]: + """An event bounded as every event is before it is queued: the message cut + to ``MAX_TELEMETRY_MESSAGE_CHARS`` and the details to + ``MAX_TELEMETRY_DETAILS_BYTES``. Its identifiers are the longest the + collector accepts.""" + + return { + "timestamp": "2026-10-09T00:00:00.000000+00:00", + "event_type": "calibration", + "stage_id": LONGEST_IDENTIFIER, + "status": "progress", + "message": ( + sanitize_text(message, limit=MAX_TELEMETRY_MESSAGE_CHARS) + if message + else None + ), + "details": sanitize_details(details), + } + + +def _assert_requests_fit_and_keep_order(service: Service, collector: FakeCollector): + """The size and order invariants, over every events request sent so far.""" + + queued = service.queued + start = 0 + for size, sequences in collector.event_requests: + assert sequences, "an events request was empty" + assert size <= COLLECTOR_BODY_LIMIT, (size, len(sequences)) + count = len(sequences) + # In order, none skipped, none repeated. + assert sequences == list(range(start + 1, start + count + 1)) + assert count <= BATCH_SIZE + if count < BATCH_SIZE and start + count < len(queued): + # It stopped short only because the next event would not fit. + one_more = {"events": queued[start : start + count + 1]} + assert len(collector_module._request_body(one_more)) > COLLECTOR_BODY_LIMIT + start += count + assert start == len(queued) + assert [event["event_id"] for event in collector.accepted] == [ + event["event_id"] for event in queued + ] + + +def test_the_limit_requests_are_cut_to_is_the_collectors() -> None: + assert MAX_EVENTS_REQUEST_BYTES == COLLECTOR_BODY_LIMIT + assert collector_module.MAX_EVENTS_REQUEST_BYTES == COLLECTOR_BODY_LIMIT + + +@pytest.mark.parametrize( + ("over", "requests"), [(0, [[1, 2]]), (1, [[1], [2]])], ids=["exact", "one-over"] +) +def test_two_events_of_exactly_the_limit_go_together_and_a_byte_more_apart( + service, collector, capsys, over, requests +) -> None: + """The collector takes a body of exactly its limit and refuses one byte + more, so that is where a request is cut.""" + service.emit() + # The second event differs from the first only in its details, so its + # padding sets the size of the request that would carry both. + first = len(collector_module._request_body(service.queued[0])) + envelope_and_comma = len('{"events":[,]}') + padding = COLLECTOR_BODY_LIMIT + over - envelope_and_comma - 2 * first + padding -= len('{"pad":""}') - len("{}") + service.emit({**SMALL_EVENT, "details": {"pad": "x" * padding}}) + both = len(collector_module._request_body({"events": service.queued})) + assert both == COLLECTOR_BODY_LIMIT + over + + service.flush_until_idle() + + assert [sequences for _, sequences in collector.event_requests] == requests + assert set(collector.asked("events")) == {202} + assert not service.spool.has_pending() + assert capsys.readouterr().err == "" + + +@pytest.mark.parametrize( + ("character", "requests"), + [ + # One byte a character: a full batch fits, and is still sent whole. + ("m", 1), + # json.dumps escapes a non-ASCII character to 6 or 12 bytes, and a full + # batch of such events is over the limit. + ("é", 2), + ("\U0001f600", 2), + ], +) +def test_a_full_batch_of_the_largest_events_is_cut_to_fit( + spool, collector, capsys, character, requests +) -> None: + event = _bounded_event( + character * 2 * MAX_TELEMETRY_MESSAGE_CHARS, _largest_details() + ) + assert len(event["message"]) == MAX_TELEMETRY_MESSAGE_CHARS + with _service(spool, collector, _new_registration(longest=True)) as service: + service.emit(event, BATCH_SIZE) + whole = len(collector_module._request_body({"events": service.queued})) + assert (whole > COLLECTOR_BODY_LIMIT) == (requests > 1), whole + + service.flush_until_idle() + + assert len(collector.event_requests) == requests + _assert_requests_fit_and_keep_order(service, collector) + assert not service.spool.has_pending() + assert service.local_only_reason() is None + assert capsys.readouterr().err == "" + + +MESSAGE = st.one_of( + st.none(), + st.tuples( + st.sampled_from(["m", "é", "\U0001f600", "\x01", '"']), + st.integers(1, MAX_TELEMETRY_MESSAGE_CHARS + 100), + ), +) +#: Keys, the character their values repeat, and how long each value is. +DETAILS = st.tuples( + st.integers(0, 60), + st.sampled_from(["x", "é", "\U0001f600"]), + st.integers(0, 2_100), +) +#: A few event shapes, and which of them each queued event takes. +SHAPES = st.lists(st.tuples(MESSAGE, DETAILS), min_size=1, max_size=5) +ORDER = st.lists(st.integers(0, 4), min_size=1, max_size=2 * BATCH_SIZE + 30) +LARGEST = (("\U0001f600", MAX_TELEMETRY_MESSAGE_CHARS), (60, "x", 2_100)) + + +@settings( + max_examples=40, + deadline=None, + suppress_health_check=[HealthCheck.function_scoped_fixture, HealthCheck.too_slow], +) +@given(shapes=SHAPES, order=ORDER) +# Every event at its largest: 100 of them are 1.4 MiB. +@example(shapes=[LARGEST], order=[0] * (2 * BATCH_SIZE + 30)) +# Large and small events alternating, so the cut falls between unequal events. +@example(shapes=[LARGEST, (None, (0, "x", 0))], order=[0, 1] * BATCH_SIZE) +def test_every_request_fits_and_events_arrive_in_order(spool, shapes, order) -> None: + """For any queue of events within those bounds: every request body is + within the collector's limit, and the collector receives every event, in + sequence order, none skipped.""" + events = [] + for message, (keys, character, length) in shapes: + text = message[0] * message[1] if message is not None else None + events.append(_bounded_event(text, _details(keys, character, length))) + collector = FakeCollector() + with _service(spool, collector, _new_registration(longest=True)) as service: + for index in order: + service.emit(events[index % len(events)]) + + service.flush_until_idle() + + assert 413 not in collector.asked("events") + _assert_requests_fit_and_keep_order(service, collector) + assert not service.spool.has_pending() + assert service.local_only_reason() is None + + +JSON = st.recursive( + st.one_of( + st.none(), + st.booleans(), + st.integers(), + st.floats(allow_nan=False, allow_infinity=False), + st.text(max_size=20), + ), + lambda children: st.one_of( + st.lists(children, max_size=4), + st.dictionaries(st.text(max_size=8), children, max_size=4), + ), + max_leaves=12, +) +EVENTS = st.lists(st.dictionaries(st.text(max_size=8), JSON, max_size=4), max_size=12) + + +@settings(max_examples=400, deadline=None) +@given(events=EVENTS, limit=st.integers(0, 400)) +@example(events=[{}, {}], limit=18) # {"events":[{},{}]} is exactly 18 bytes +@example(events=[{}, {}], limit=17) +def test_the_events_sent_are_the_longest_leading_run_that_fits(events, limit) -> None: + """For any events and any limit, against the body that would be sent.""" + + def body(count: int) -> int: + return len(collector_module._request_body({"events": events[:count]})) + + with mock.patch.object(collector_module, "MAX_EVENTS_REQUEST_BYTES", limit): + fitting = collector_module._events_fitting_request(events) + + count = len(fitting) + assert fitting == events[:count] + assert bool(fitting) == bool(events) + assert count <= 1 or body(count) <= limit + assert count == len(events) or body(count + 1) > limit + # Only an event that cannot fit alone is ever sent over the limit. + if fitting and body(count) > limit: + assert count == 1 and body(1) > limit + + +def test_the_bytes_measured_are_the_bytes_sent() -> None: + """``_http_post`` sends exactly what ``_request_body`` returns, and says so + in Content-Length, which is what the collector's limit reads.""" + received: list[tuple[str, bytes]] = [] + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + length = self.headers["Content-Length"] + received.append((length, self.rfile.read(int(length)))) + self.send_response(202) + self.send_header("Content-Length", "2") + self.end_headers() + self.wfile.write(b"{}") + + def log_message(self, format, *args): + return + + payload = { + "events": [ + {"message": 'café \U0001f600 \x01 "quoted"', "details": {"n": 1.5}}, + {"message": None, "details": {}}, + ] + } + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + server_thread = threading.Thread(target=server.serve_forever) + server_thread.start() + try: + status, _ = collector_module._http_post( + f"http://127.0.0.1:{server.server_port}/v1/runs/run-a/events", + payload, + "collector-token", + ) + finally: + server.shutdown() + server_thread.join(timeout=2) + server.server_close() + + body = collector_module._request_body(payload) + assert status == 202 + assert received == [(str(len(body)), body)] + assert json.loads(body) == payload + assert len(body) == ( + collector_module._EVENTS_ENVELOPE_BYTES + + sum(len(collector_module._request_body(e)) for e in payload["events"]) + + len(payload["events"]) + - 1 + ) + + +def _oversized_event() -> dict[str, Any]: + """An event no request can carry. Every event is bounded before this + version queues it, but the spool is shared with other versions and holds + whatever they queued.""" + + return {**SMALL_EVENT, "details": {"blob": "x" * COLLECTOR_BODY_LIMIT}} + + +def test_an_event_too_large_to_send_alone_is_left_to_the_collector( + service, collector, capsys +) -> None: + """The events queued before it are delivered first, in a request of their + own. It is then sent alone, and the collector's 413 settles the run.""" + service.emit(count=3) + service.emit(_oversized_event()) + service.emit(count=2) + + service.flush_until_idle() + + assert [sequences for _, sequences in collector.event_requests] == [[1, 2, 3], [4]] + assert collector.event_requests[1][0] > COLLECTOR_BODY_LIMIT + assert collector.asked("events") == [202, 413] + assert [e["sequence"] for e in collector.accepted] == [1, 2, 3] + assert service.local_only_reason() == LOCAL_ONLY_REJECTED_EVENTS + assert capsys.readouterr().err == _expected_warning("events", 413) + # Local-only is final: nothing more is sent, however long the service waits. + service.wait_out_any_retry_delay() + assert not service.flush() + assert len(collector.event_requests) == 2 + + +def test_an_event_over_the_limit_that_the_collector_takes_does_not_stop_the_run( + service, collector, capsys +) -> None: + """The limit is the collector's to enforce: were it raised there, an event + over this service's figure would be acknowledged like any other.""" + collector.body_limit = 4 * COLLECTOR_BODY_LIMIT + service.emit() + service.emit(_oversized_event()) + service.emit(count=2) + + service.flush_until_idle() + + assert [sequences for _, sequences in collector.event_requests] == [ + [1], + [2], + [3, 4], + ] + assert collector.asked("events") == [202, 202, 202] + assert not service.spool.has_pending() + assert service.local_only_reason() is None + assert capsys.readouterr().err == "" + + +# --- A run the collector no longer has --------------------------------------- + + +def test_a_run_the_collector_lost_is_registered_again_and_delivered( + service, collector, capsys +) -> None: + service.emit(count=2) + assert service.flush() + service.emit(count=3) + collector.runs.clear() # the collector's database was restored, say + + assert not service.flush() + # Nothing is decided yet, and nothing waits on a retry delay. + assert service.pending() and service.local_only_reason() is None + assert service.flush() + + assert collector.answers == [ + ("exchange", 200), + ("register", 201), + ("events", 202), + ("events", 404), + ("register", 201), + ("events", 202), + ] + # The batch the collector did not find is the batch it is then sent. + assert collector.event_requests[1][1] == collector.event_requests[2][1] == [3, 4, 5] + assert [e["event_id"] for e in collector.accepted] == [ + e["event_id"] for e in service.queued + ] + assert not service.spool.has_pending() + assert capsys.readouterr().err == "" + + +def test_a_run_the_collector_cannot_keep_goes_local_only_after_one_more_try( + service, collector, capsys +) -> None: + """A collector that answers 201 for the registration and 404 for the events + (an older one with no such route, say) is asked twice, then left alone.""" + collector.keeps_runs = False + service.emit() + + for _ in range(5): + assert not service.flush() + service.wait_out_any_retry_delay() + + assert collector.answers == [ + ("exchange", 200), + ("register", 201), + ("events", 404), + ("register", 201), + ("events", 404), + ] + assert service.local_only_reason() == LOCAL_ONLY_REJECTED_EVENTS + assert capsys.readouterr().err == _expected_warning("events", 404) + + +def test_each_acknowledged_batch_allows_one_more_registration( + service, collector, capsys +) -> None: + """The collector loses the run three times, with a batch acknowledged + between each: every loss is recovered.""" + service.emit() + assert service.flush() + for _ in range(3): + service.emit() + collector.runs.clear() + assert not service.flush() + assert service.flush() + + assert collector.asked("register") == [201, 201, 201, 201] + assert collector.asked("events") == [202, 404, 202, 404, 202, 404, 202] + assert not service.spool.has_pending() + assert service.local_only_reason() is None + assert capsys.readouterr().err == "" + + +@pytest.mark.parametrize("status", [403, 409, 422]) +def test_registering_again_is_answered_like_any_registration( + service, collector, capsys, status +) -> None: + """Another user may have registered the run in the meantime (403), or the + owner with other metadata (409): the second registration settles the run + as the first would have.""" + service.emit() + assert service.flush() + service.emit() + collector.runs.clear() + assert not service.flush() + collector.refuse["register"] = status + + for _ in range(3): + assert not service.flush() + service.wait_out_any_retry_delay() + + assert collector.asked("register") == [201, status] + assert collector.asked("events") == [202, 404] + assert service.local_only_reason() == LOCAL_ONLY_REJECTED_REGISTRATION + assert capsys.readouterr().err == _expected_warning("register", status) + + +@pytest.mark.parametrize("status", [503, 429, 408]) +def test_registering_again_waits_out_a_transient_answer( + service, collector, capsys, status +) -> None: + service.emit() + assert service.flush() + service.emit() + collector.runs.clear() + assert not service.flush() + collector.refuse["register"] = status + + assert not service.flush() + assert not service.flush() # inside the retry delay: nothing is sent + assert collector.asked("register") == [201, status] + del collector.refuse["register"] + service.wait_out_any_retry_delay() + assert service.flush() + + assert collector.asked("register") == [201, status, 201] + assert collector.asked("events") == [202, 404, 202] + assert not service.spool.has_pending() + assert capsys.readouterr().err == "" + + +def test_no_other_answer_to_events_registers_the_run_again(spool) -> None: + """Over every status but 404: a run is registered once.""" + for status in [s for s in range(200, 600) if s not in (202, 404)]: + collector = FakeCollector() + collector.refuse["events"] = status + with _service(spool, collector) as service: + service.emit() + for _ in range(4): + assert not service.flush(), status + service.wait_out_any_retry_delay() + assert collector.asked("register") == [201], status + + +def _assert_registrations_are_bounded(collector: FakeCollector) -> None: + """Between two registrations the collector accepted there is a 404 for the + run's events; from the second on, an acknowledged batch as well.""" + + since: list[int] = [] + accepted_registrations = 0 + for kind, status in collector.answers: + if kind == "events": + since.append(status) + elif kind == "register" and status == 201: + if accepted_registrations >= 1: + assert 404 in since, collector.answers + if accepted_registrations >= 2: + assert 202 in since, collector.answers + accepted_registrations += 1 + since = [] + + +STEPS = st.lists( + st.sampled_from( + ["emit", "flush", "flush", "flush", "lose", "stop_keeping", "keep_again"] + ), + max_size=40, +) + + +@settings( + max_examples=200, + deadline=None, + suppress_health_check=[HealthCheck.function_scoped_fixture, HealthCheck.too_slow], +) +@given(steps=STEPS) +@example(steps=["emit", "flush", "emit", "lose", "flush", "lose", "flush", "flush"]) +@example(steps=["stop_keeping", "emit", "flush", "flush", "keep_again", "flush"]) +def test_a_lost_run_costs_a_bounded_number_of_registrations(spool, steps) -> None: + """The collector loses the run, or stops keeping registrations, at any + point. After every step: registrations stay bounded as above; a run goes + local-only only on a second 404 with nothing acknowledged since the first; + nothing is sent after that; and what the collector acknowledged is the + queue in order, none skipped or repeated. In the end, a collector that + keeps the run again receives everything, unless the run went local-only.""" + collector = FakeCollector() + with _service(spool, collector) as service: + settled_at: int | None = None + + def check() -> None: + nonlocal settled_at + _assert_registrations_are_bounded(collector) + accepted = [event["event_id"] for event in collector.accepted] + queued = [event["event_id"] for event in service.queued] + assert accepted == queued[: len(accepted)] + reason = service.local_only_reason() + if settled_at is not None: + assert len(collector.answers) == settled_at, "sent after local-only" + elif reason is not None: + assert reason == LOCAL_ONLY_REJECTED_EVENTS + events = collector.asked("events") + assert events[-1] == 404 + # The 404 before it, with only the registration in between. + assert collector.answers[-3:] == [ + ("events", 404), + ("register", 201), + ("events", 404), + ] + settled_at = len(collector.answers) + + for step in steps: + if step == "emit": + service.emit() + elif step == "flush": + service.flush() + elif step == "lose": + collector.runs.clear() + else: + collector.keeps_runs = step == "keep_again" + check() + + collector.keeps_runs = True + for _ in range(4): + service.flush() + check() + if settled_at is None: + assert not service.spool.run_has_deliverable( + service.registration["run_id"], service.registration["producer_id"] + ) + assert len(collector.accepted) == len(service.queued) diff --git a/tests/telemetry/test_delivery.py b/tests/telemetry/test_delivery.py index b4d2b07..4e8dd0d 100644 --- a/tests/telemetry/test_delivery.py +++ b/tests/telemetry/test_delivery.py @@ -317,7 +317,6 @@ def reject(url, payload, token, *, timeout=5.0): ("status", "retried"), [ (422, False), - (404, False), (413, False), (400, False), (429, True), @@ -330,7 +329,8 @@ def test_a_settled_collector_rejection_goes_local_only_instead_of_retrying( ) -> None: """A 4xx the collector will repeat (an event shape it does not accept, say) must not wedge the queue: the run goes local-only with a reason and one - warning. A timeout, a rate limit or a server error still retries.""" + warning. A timeout, a rate limit or a server error still retries. A 404 is + first answered by registering the run again: see ``test_collector_delivery``.""" spool_path = tmp_path / "events.sqlite3" spool = EventSpool(spool_path) registration = _registration() @@ -455,7 +455,8 @@ def test_every_client_error_has_one_outcome( ) -> None: """Every 4xx, on either request. 401 is about the collector token, which is exchanged again; 403 is about the login; 408 and 429 wait and retry; any - other is the collector's settled answer about the run.""" + other is the collector's settled answer about the run. A 404 for a run's + events settles it the second time: the first has the run registered again.""" spool_path = tmp_path / "events.sqlite3" spool = EventSpool(spool_path) monkeypatch.setattr(collector_module, "_huggingface_token", lambda: "hf-member") @@ -512,7 +513,12 @@ def fake_post(url, payload, token, *, timeout=5.0, status=status, sent=sent): assert reason is None, status assert warnings == "", status continue - assert again == [], (status, again) + if refused == "events" and status == 404: + # Registered again and sent again, once. + assert again == [RUN_REGISTRATION_PATH, refused_path], again + first = len(sent) + else: + assert again == [], (status, again) if status in (408, 429): assert reason is None, status assert warnings == "", status @@ -522,8 +528,8 @@ def fake_post(url, payload, token, *, timeout=5.0, status=status, sent=sent): continue assert reason == (login_reason if status == 403 else settled_reason), status assert warnings.count("local-only for this run") == 1, status - if refused == "events" and status != 403: - assert f"HTTP {status}" in warnings, status + assert f"HTTP {status}" in warnings, status + assert "organization member" not in warnings, status # Nothing is sent again, however long the service waits. clock[0] += MAX_RETRY_SECONDS + 1 assert not delivery.flush_once(), status diff --git a/tests/telemetry/test_foreign_runs.py b/tests/telemetry/test_foreign_runs.py index eabd079..322ef3e 100644 --- a/tests/telemetry/test_foreign_runs.py +++ b/tests/telemetry/test_foreign_runs.py @@ -8,7 +8,8 @@ Hugging Face user to register a run owns it and anyone else gets 403; the owner registering it again with other metadata gets 409; a registration or a batch outside the collector's schema gets 422, whoever sends it; and an unknown -run's events get 404. +run's events get 404. The fake can also lose a run it registered, as a restored +database would, and can be told never to find a run's events. Invariants (each checked below; the first four also after every step of a property test that runs services in any order and lets one run while @@ -17,7 +18,8 @@ - No service makes another producer's run local-only because of its own credential (missing, refused, or not the run's owner). Another producer's run goes local-only only after a collector answer about the run itself: a - registration 409, or a registration or batch refused with 422. + registration 409, a registration or batch refused with 422, or a second 404 + for its events with no batch acknowledged to that service since the first. - While a producer's service is alive, no other service sends any request for that producer's run. - A run goes local-only only when the deciding service's latest collector @@ -29,7 +31,8 @@ - Once every service has gone, a run left pending is delivered in full by a service whose credential owns it, including a service that was refused for the run under an earlier login, unless the collector refuses the run itself - (409 or 422), which makes it local-only. + (409, 422, or never finding it), which makes it local-only. If the collector + lost the run, the first service to register it again owns it. """ from __future__ import annotations @@ -66,6 +69,7 @@ from microcosm_emitter.telemetry.service import leases as leases_module from microcosm_emitter.telemetry.service.collector import CollectorDelivery from microcosm_emitter.telemetry.service.constants import ( + BATCH_SIZE, DEFAULT_TOKEN_LIFETIME_SECONDS, LOCAL_ONLY_MISSING_CREDENTIAL, LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION, @@ -124,9 +128,12 @@ def __init__( rejected_run_ids: set[str] | None = None, conflicted_run_ids: set[str] | None = None, malformed_run_ids: set[str] | None = None, + unfindable_run_ids: set[str] | None = None, ) -> None: # Runs whose batches are outside the collector's schema. self.rejected_run_ids = set(rejected_run_ids or ()) + # Runs it registers and then never finds when their events arrive. + self.unfindable_run_ids = set(unfindable_run_ids or ()) # Runs whose registration is outside it. The schema is checked before # the run's owner, so nobody comes to own such a run. self.malformed_run_ids = set(malformed_run_ids or ()) @@ -192,7 +199,7 @@ def post(self, caller: Run, url: str, payload, token: str) -> tuple[int, dict]: run_id = path.split("/")[3] events = payload["events"] run = (run_id, events[0]["producer_id"]) - if run_id not in self.owners: + if run_id not in self.owners or run_id in self.unfindable_run_ids: return self._answer(caller, "events", run, 404, {}) if self.owners[run_id] != user or run not in self.producers: return self._answer(caller, "events", run, 403, {}) @@ -205,6 +212,13 @@ def _answer(self, caller, kind, run, status, body): self.requests.append((caller, kind, run, status)) return status, body + def forget(self, run_id: str) -> None: + """Lose a registered run. What was acknowledged stays on record here.""" + + self.owners.pop(run_id, None) + self.metadata.pop(run_id, None) + self.producers = {run for run in self.producers if run[0] != run_id} + def latest(self, caller: Run, before: int) -> tuple[str, Run | None, int] | None: """The caller's last request among the first ``before``: kind, run, status.""" @@ -373,6 +387,23 @@ def host(tmp_path): host.close() +def _assert_events_were_refused(host: Host, caller: Run, run: Run, at: int) -> None: + """The service's latest answer refused this run's events: with 422, or with + a second 404 and no batch of the run acknowledged to it since the first.""" + + latest = host.collector.latest(caller, at) + if latest == ("events", run, 422): + assert run[0] in host.collector.rejected_run_ids, (caller, run) + return + assert latest == ("events", run, 404), (caller, run, latest) + answers = [ + status + for who, kind, about, status in host.collector.requests[:at] + if (who, kind, about) == (caller, "events", run) + ] + assert answers[-2:] == [404, 404], (caller, run, answers) + + def _assert_own_credential_only(host: Host) -> None: """The safety invariants, over everything recorded so far.""" @@ -388,8 +419,7 @@ def _assert_own_credential_only(host: Host) -> None: # Another producer's run: only the collector's answer about the # run itself, never one about this service's credential. if reason == LOCAL_ONLY_REJECTED_EVENTS: - assert run[0] in host.collector.rejected_run_ids, (caller, run) - assert latest == ("events", run, 422), (caller, run, latest) + _assert_events_were_refused(host, caller, run, at) continue assert reason == LOCAL_ONLY_REJECTED_REGISTRATION, (caller, run, reason) if run[0] in host.collector.malformed_run_ids: @@ -415,10 +445,7 @@ def _assert_own_credential_only(host: Host) -> None: assert latest == ("events", run, 403), (run, latest) else: assert reason == LOCAL_ONLY_REJECTED_EVENTS, reason - assert latest in (("events", run, 404), ("events", run, 422)), ( - run, - latest, - ) + _assert_events_were_refused(host, caller, run, at) # A refused credential is exchanged once per service, not on every pass. refused = host.collector.refused assert len(set(refused)) == len(refused), refused @@ -764,6 +791,8 @@ def hold(): # run then keeps, and every status sent about it. RUN_REFUSALS = { "unacceptable_events": (LOCAL_ONLY_REJECTED_EVENTS, [201, 422]), + # Not found once, registered again, not found again. + "unfindable_run": (LOCAL_ONLY_REJECTED_EVENTS, [201, 404, 201, 404]), "unacceptable_registration": (LOCAL_ONLY_REJECTED_REGISTRATION, [422]), "registration_conflict": (LOCAL_ONLY_REJECTED_REGISTRATION, [409]), } @@ -772,6 +801,8 @@ def hold(): def _refuse_run(host: Host, run: Run, answer: str) -> None: if answer == "unacceptable_events": host.collector.rejected_run_ids.add(run[0]) + elif answer == "unfindable_run": + host.collector.unfindable_run_ids.add(run[0]) elif answer == "unacceptable_registration": host.collector.malformed_run_ids.add(run[0]) else: @@ -786,7 +817,8 @@ def test_an_orphan_the_collector_refuses_goes_local_only_without_a_warning_here( ): """A 409 reaches only the run's owner, and every owner would get it; a 422 is about the registration or the batch, which is the same whoever sends - it.""" + it; and a run the collector registers but does not then find is not found + by anyone.""" reason, statuses = RUN_REFUSALS[answer] build_a = host.start(MEMBER + "alice") @@ -800,7 +832,7 @@ def test_an_orphan_the_collector_refuses_goes_local_only_without_a_warning_here( host.flush(build_b) assert host.states()[build_a.run] == ("local_only", reason) - # One refusal settles it: the run is neither retried nor left queued for + # The refusal settles it: the run is neither retried nor left queued for # another service. assert [ status @@ -818,7 +850,8 @@ def test_an_orphan_the_collector_refuses_goes_local_only_without_a_warning_here( @pytest.mark.parametrize("answer", RUN_REFUSALS) def test_a_refused_orphan_does_not_hold_back_the_adopters_own_run(host, answer): """A settled refusal is not retried, so it sets no retry delay; a delay - would gate every run the service delivers, its own included.""" + would gate every run the service delivers, its own included. Nor does the + 404 that has a run registered again.""" build_a = host.start(MEMBER + "alice") host.emit(build_a) @@ -827,12 +860,71 @@ def test_a_refused_orphan_does_not_hold_back_the_adopters_own_run(host, answer): build_b = host.start(MEMBER + "alice") host.emit(build_b) host.flush(build_b) + settled_at_once = answer != "unfindable_run" + assert (host.states()[build_a.run][0] == "local_only") is settled_at_once + + for _ in range(2): + host.emit(build_b) + host.flush(build_b, advance=False) + assert host.collector.accepted[build_b.run] == set(build_b.appended) + assert host.states()[build_a.run][0] == "local_only" + _assert_own_credential_only(host) + + +def test_an_adopted_orphan_the_collector_loses_is_registered_again(host, capsys): + """An adopter is answered as the run's own service would be: after a 404 + it registers the orphan again and delivers the rest.""" + + build_a = host.start(MEMBER + "alice") + host.emit(build_a, BATCH_SIZE + 1) + host.kill(build_a) + build_b = host.start(MEMBER + "alice") + + assert host.flush(build_b) + host.collector.forget(build_a.run[0]) + assert not host.flush(build_b) + assert host.states()[build_a.run] == ("pending", None) + assert host.flush(build_b) + + assert host.collector.asked(build_b.run, "register", build_a.run) == [201, 201] + assert host.collector.asked(build_b.run, "events", build_a.run) == [202, 404, 202] + assert host.collector.accepted[build_a.run] == set(build_a.appended) + assert host.queued(build_a.run) == 0 + assert "local-only" not in capsys.readouterr().err + _assert_own_credential_only(host) + +def test_a_lost_run_another_user_registers_first_is_refused_its_first_owner( + host, capsys +): + """The collector gives a run it does not have to whoever registers it. So + registering again can be refused, like any registration: only for this + login, and only this build's own run.""" + + build_a = host.start(MEMBER + "alice", run_id="shared") + host.emit(build_a) + host.flush(build_a) + host.emit(build_a) + host.collector.forget("shared") + host.flush(build_a) + build_b = host.start(MEMBER + "bob", run_id="shared") host.emit(build_b) - host.flush(build_b, advance=False) + host.flush(build_b) + + host.flush(build_a) + assert host.collector.asked(build_a.run, "register", build_a.run) == [201, 403] + assert host.states()[build_a.run] == ( + "local_only", + LOCAL_ONLY_REJECTED_REGISTRATION, + ) + assert host.states()[build_b.run] == ("pending", None) assert host.collector.accepted[build_b.run] == set(build_b.appended) + warning = capsys.readouterr().err + assert warning.count("local-only for this run") == 1 + assert "Hugging Face user" in warning and "HTTP 403" in warning + assert "organization member" not in warning _assert_own_credential_only(host) @@ -1318,9 +1410,14 @@ def log_message(self, format, *args): def _label_paths(host: Host) -> None: """Report which delivery paths a generated scenario reached.""" + registered: set[tuple[Run, Run]] = set() for caller, kind, about, status in host.collector.requests: if about is not None and about != caller: event(f"adopted orphan: {kind} {status}") + if kind == "register" and status == 201: + if (caller, about) in registered: + event(f"{'own' if caller == about else 'adopted'} run registered again") + registered.add((caller, about)) for caller, run, reason, *_ in host.local_only: event(f"{'own' if caller == run else 'adopted'} run local-only: {reason}") @@ -1338,6 +1435,8 @@ def _label_paths(host: Host) -> None: st.tuples(st.just("flush"), PRODUCER), st.tuples(st.just("kill"), PRODUCER), st.tuples(st.just("login"), PRODUCER, st.sampled_from(CREDENTIALS)), + # The collector loses a run it registered, any build's. + st.tuples(st.just("forget"), PRODUCER), # One service flushes while another's request is in flight, and may # then exit: the first service's list of pending runs goes stale. st.tuples(st.just("interleave"), PRODUCER, PRODUCER, st.booleans()), @@ -1368,6 +1467,7 @@ def _start_build(host: Host, build: tuple[str | None, int]) -> Producer: rejected=st.sets(st.integers(0, 7), max_size=2), conflicted=st.sets(st.integers(0, 7), max_size=2), malformed=st.sets(st.integers(0, 7), max_size=1), + unfindable=st.sets(st.integers(0, 7), max_size=1), # The logins the two last services start under, before Alice's and Bob's. rescue_logins=st.tuples(st.sampled_from(CREDENTIALS), st.sampled_from(CREDENTIALS)), ) @@ -1380,6 +1480,7 @@ def _start_build(host: Host, build: tuple[str | None, int]) -> Producer: rejected=set(), conflicted=set(), malformed=set(), + unfindable=set(), rescue_logins=(MEMBER + "alice", MEMBER + "bob"), ) # From review: a service refused for every pending run, whose own run is not @@ -1390,10 +1491,30 @@ def _start_build(host: Host, build: tuple[str | None, int]) -> Producer: rejected=set(), conflicted=set(), malformed=set(), + unfindable=set(), rescue_logins=(OUTSIDER, EXPIRED), ) +# Alice's service loses its run at the collector, and Bob's build of the same +# run id registers it first. +@example( + builds=[(MEMBER + "alice", 0), (MEMBER + "bob", 1)], + operations=[("flush", 0), ("emit", 0, 1), ("forget", 0), ("flush", 0)] + + [("flush", 1), ("flush", 0)], + rejected=set(), + conflicted=set(), + malformed=set(), + unfindable=set(), + rescue_logins=(MEMBER + "alice", MEMBER + "bob"), +) def test_no_service_decides_another_producers_run_from_its_own_credential( - tmp_path_factory, builds, operations, rejected, conflicted, malformed, rescue_logins + tmp_path_factory, + builds, + operations, + rejected, + conflicted, + malformed, + unfindable, + rescue_logins, ): host = Host( tmp_path_factory.mktemp("foreign"), @@ -1401,6 +1522,7 @@ def test_no_service_decides_another_producers_run_from_its_own_credential( {f"run-{index}" for index in rejected}, {f"run-{index}" for index in conflicted}, {f"run-{index}" for index in malformed}, + {f"run-{index}" for index in unfindable}, ), ) try: @@ -1412,6 +1534,10 @@ def test_no_service_decides_another_producers_run_from_its_own_credential( if kind == "start": _start_build(host, operation[1]) continue + if kind == "forget": + lost = host.producers[operation[1] % len(host.producers)] + host.collector.forget(lost.run[0]) + continue live = [producer for producer in host.producers if producer.alive] if not live: continue @@ -1477,8 +1603,12 @@ def other_flushes(other=other, exits=exits) -> None: LOCAL_ONLY_REJECTED_REGISTRATION, ), producer.run continue - elif run_id in host.collector.rejected_run_ids: - # Its owner, whoever that is, gets 422: a refusal of the batch. + elif ( + run_id in host.collector.rejected_run_ids + or run_id in host.collector.unfindable_run_ids + ): + # Its owner, whoever that is, gets 422, or 404 twice: a + # refusal of the batch. assert after[producer.run] == ( "local_only", LOCAL_ONLY_REJECTED_EVENTS, diff --git a/tests/test_collector_contract.py b/tests/test_collector_contract.py index 51afe67..193e1c4 100644 --- a/tests/test_collector_contract.py +++ b/tests/test_collector_contract.py @@ -12,11 +12,23 @@ from sqlalchemy.orm import Session from microcosm_emitter.telemetry.client import LocalTelemetryEmitter, TelemetryRun +from microcosm_emitter.telemetry.protocol import ( + MAX_TELEMETRY_DETAILS_BYTES, + MAX_TELEMETRY_MESSAGE_CHARS, +) from microcosm_emitter.telemetry.service import collector as delivery_module from microcosm_emitter.telemetry.service.collector import CollectorDelivery -from microcosm_emitter.telemetry.service.constants import LOCAL_ONLY_REJECTED_EVENTS +from microcosm_emitter.telemetry.service.constants import ( + BATCH_SIZE, + LOCAL_ONLY_REJECTED_EVENTS, + LOCAL_ONLY_REJECTED_REGISTRATION, + MAX_EVENTS_REQUEST_BYTES, + RUN_REGISTRATION_PATH, + TOKEN_EXCHANGE_PATH, +) from microcosm_emitter.telemetry.service.database import create_spool_engine from microcosm_emitter.telemetry.service.models import TelemetryRunRecord +from microcosm_emitter.telemetry.service.sanitization import sanitize_details from microcosm_emitter.telemetry.service.spool import EventSpool pytestmark = pytest.mark.collector @@ -42,11 +54,19 @@ def live_collector(): class FakeIdentityProvider: def authenticate(self, token, required_org): - assert token in {"hf_test_member", "hf_test_outsider"} + assert token in { + "hf_test_member", + "hf_test_other_member", + "hf_test_outsider", + } return HuggingFacePrincipal( - user_id="test-user", + user_id=( + "other-test-user" + if token == "hf_test_other_member" + else "test-user" + ), username="fixture", - organizations=(required_org,) if token == "hf_test_member" else (), + organizations=() if token == "hf_test_outsider" else (required_org,), ) upgrade_database(database_url) @@ -204,15 +224,44 @@ def lose_ack(url, *args, **kwargs): ] -@pytest.mark.parametrize( - ("refusal", "status"), - [("event_outside_schema", 422), ("run_not_registered", 404)], -) -def test_a_settled_refusal_makes_the_run_local_only_instead_of_retrying( - tmp_path, monkeypatch, capsys, live_collector, refusal, status +def _local_only_reason(path, run: TelemetryRun) -> str | None: + engine = create_spool_engine(path) + try: + with Session(engine) as session: + stored = session.get(TelemetryRunRecord, (run.run_id, run.producer_id)) + return stored.local_only_reason + finally: + engine.dispose() + + +def _recording_post(monkeypatch) -> list[tuple[str, int, int]]: + """Record every request the delivery sends: path, body bytes, status.""" + + real_post = delivery_module._http_post + sent: list[tuple[str, int, int]] = [] + + def recording_post(url, payload, *args, **kwargs): + response = real_post(url, payload, *args, **kwargs) + size = len(delivery_module._request_body(payload)) + sent.append((urlsplit(url).path, size, response[0])) + return response + + monkeypatch.setattr(delivery_module, "_http_post", recording_post) + return sent + + +def _collector_token(address: str, hf_token: str) -> str: + status, body = delivery_module._http_post( + address + TOKEN_EXCHANGE_PATH, {}, hf_token + ) + assert status == 200, body + return body["access_token"] + + +def test_an_event_outside_the_schema_makes_the_run_local_only_instead_of_retrying( + tmp_path, monkeypatch, capsys, live_collector ): - """The collector's own answers to a batch it will never accept: 422 for an - event outside its schema, 404 for a run it does not know.""" + """The collector's own answer to a batch it will never accept: 422.""" address, reader = live_collector monkeypatch.setenv("HF_TOKEN", "hf_test_member") run = TelemetryRun(uuid.uuid4().hex, "UK", "contract") @@ -220,44 +269,238 @@ def test_a_settled_refusal_makes_the_run_local_only_instead_of_retrying( path = tmp_path / "events.sqlite3" spool = EventSpool(path) spool.register(registration) - event_type = "not_a_collector_event_type" if status == 422 else "run" - spool.append(registration, {"event_type": event_type, "status": "started"}) - real_post = delivery_module._http_post - answers: list[tuple[str, int]] = [] - - def recording_post(url, *args, **kwargs): - response = real_post(url, *args, **kwargs) - answers.append((urlsplit(url).path, response[0])) - return response - - monkeypatch.setattr(delivery_module, "_http_post", recording_post) + spool.append( + registration, {"event_type": "not_a_collector_event_type", "status": "started"} + ) + sent = _recording_post(monkeypatch) delivery = CollectorDelivery(spool, registration, development_collector_url=address) - if refusal == "run_not_registered": - # As if the collector no longer had a run this service registered. - delivery._registered.add((run.run_id, run.producer_id)) events_path = f"/v1/runs/{run.run_id}/events" try: assert not delivery.flush_once() assert not delivery.flush_once() - assert [code for sent, code in answers if sent == events_path] == [status] + assert [code for where, _, code in sent if where == events_path] == [422] assert spool.has_pending() and not spool.has_deliverable() finally: delivery.close() spool.close() - engine = create_spool_engine(path) - try: - with Session(engine) as session: - stored = session.get(TelemetryRunRecord, (run.run_id, run.producer_id)) - assert stored.local_only_reason == LOCAL_ONLY_REJECTED_EVENTS - finally: - engine.dispose() + assert _local_only_reason(path, run) == LOCAL_ONLY_REJECTED_EVENTS warnings = capsys.readouterr().err assert warnings.count("local-only for this run") == 1 - assert f"HTTP {status}" in warnings + assert "HTTP 422" in warnings # The collector kept none of the refused batch. document = reader.get(f"/v1/runs/{run.run_id}") - if refusal == "run_not_registered": + assert document.status_code == 200, document.text + assert document.json()["events"] == [] + + +def test_a_run_the_collector_does_not_have_is_registered_again_and_delivered( + tmp_path, monkeypatch, capsys, live_collector +): + """The collector answers 404 for the events of a run it does not have. The + service then registers the run again, which the collector accepts, and the + same batch arrives.""" + address, reader = live_collector + monkeypatch.setenv("HF_TOKEN", "hf_test_member") + run = TelemetryRun(uuid.uuid4().hex, "UK", "contract") + registration = run.as_registration() + path = tmp_path / "events.sqlite3" + spool = EventSpool(path) + spool.register(registration) + event = spool.append(registration, {"event_type": "run", "status": "started"}) + sent = _recording_post(monkeypatch) + delivery = CollectorDelivery(spool, registration, development_collector_url=address) + # As if the collector no longer had a run this service registered. + delivery._registered.add(delivery_module._registration_key(registration)) + try: + assert not delivery.flush_once() + assert delivery.flush_once() + assert not spool.has_pending() + finally: + delivery.close() + spool.close() + assert [(where, code) for where, _, code in sent] == [ + (TOKEN_EXCHANGE_PATH, 200), + (f"/v1/runs/{run.run_id}/events", 404), + (RUN_REGISTRATION_PATH, 201), + (f"/v1/runs/{run.run_id}/events", 202), + ] + assert _local_only_reason(path, run) is None + assert capsys.readouterr().err == "" + document = reader.get(f"/v1/runs/{run.run_id}") + assert document.status_code == 200, document.text + assert [entry["event_id"] for entry in document.json()["events"]] == [ + event["event_id"] + ] + + +@pytest.mark.parametrize( + ("refusal", "status", "says"), + [ + ("owner_with_other_metadata", 409, "rejected its registration (HTTP 409)"), + ("outside_schema", 422, "rejected its registration (HTTP 422)"), + ("another_users_run", 403, "refused the ambient Hugging Face user"), + ], +) +def test_a_refused_registration_warns_about_the_registration_not_the_credential( + tmp_path, monkeypatch, capsys, live_collector, refusal, status, says +): + """The collector's own answers to a registration: 409 to the run's owner + for other metadata, 422 outside its schema, 403 for another user's run. In + each the token exchange had accepted the login as an organization member.""" + address, reader = live_collector + monkeypatch.setenv("HF_TOKEN", "hf_test_member") + run_id = uuid.uuid4().hex + # A lower-case country code is outside the collector's schema. + run = TelemetryRun( + run_id, "uk" if refusal == "outside_schema" else "UK", "contract" + ) + registration = run.as_registration() + if refusal != "outside_schema": + # Someone registered the run id first: this user with another + # pipeline, or another member with the same registration. + another_user = refusal == "another_users_run" + seeded = { + **registration, + "producer_id": uuid.uuid4().hex, + "pipeline": "contract" if another_user else "another-pipeline", + } + token = _collector_token( + address, "hf_test_other_member" if another_user else "hf_test_member" + ) + answer = delivery_module._http_post( + address + RUN_REGISTRATION_PATH, seeded, token + ) + assert answer[0] == 201, answer + path = tmp_path / "events.sqlite3" + spool = EventSpool(path) + spool.register(registration) + spool.append(registration, {"event_type": "run", "status": "started"}) + sent = _recording_post(monkeypatch) + delivery = CollectorDelivery(spool, registration, development_collector_url=address) + try: + assert not delivery.flush_once() + assert not delivery.flush_once() + assert [(where, code) for where, _, code in sent] == [ + (TOKEN_EXCHANGE_PATH, 200), + (RUN_REGISTRATION_PATH, status), + ] + assert spool.has_pending() and not spool.has_deliverable() + finally: + delivery.close() + spool.close() + assert _local_only_reason(path, run) == LOCAL_ONLY_REJECTED_REGISTRATION + warnings = capsys.readouterr().err + assert warnings.count("local-only for this run") == 1 + assert says in warnings and f"HTTP {status}" in warnings + assert "organization member" not in warnings + # None of the run's events reached the collector. + document = reader.get(f"/v1/runs/{run_id}") + if refusal == "outside_schema": assert document.status_code == 404 else: assert document.status_code == 200, document.text assert document.json()["events"] == [] + + +def _largest_event() -> dict: + """The largest event the service queues: a message of the most characters, + each one json.dumps escapes to twelve bytes, and details at their limit.""" + + details = {} + while True: + grown = {**details, f"k{len(details)}": "x" * 190} + if len(delivery_module._request_body(grown)) > MAX_TELEMETRY_DETAILS_BYTES - 16: + break + details = grown + padding = ( + MAX_TELEMETRY_DETAILS_BYTES + - len(delivery_module._request_body(details)) + - len(',"pad":""') + ) + details["pad"] = "x" * padding + assert len(delivery_module._request_body(details)) == MAX_TELEMETRY_DETAILS_BYTES + assert sanitize_details(details) == details + return { + "event_type": "stage", + "stage_id": "s" * 192, + "status": "progress", + "message": "\U0001f600" * MAX_TELEMETRY_MESSAGE_CHARS, + "details": details, + } + + +def test_a_full_batch_of_the_largest_events_arrives_in_requests_the_collector_takes( + tmp_path, monkeypatch, capsys, live_collector +): + """A hundred of the largest events are over the collector's 1 MiB body + limit together. They are sent in two requests, and all of them arrive.""" + address, reader = live_collector + monkeypatch.setenv("HF_TOKEN", "hf_test_member") + run = TelemetryRun("r" + uuid.uuid4().hex + "r" * 159, "UK", "contract") + assert len(run.run_id) == 192 + registration = run.as_registration() + path = tmp_path / "events.sqlite3" + spool = EventSpool(path) + spool.register(registration) + queued = spool.append_many(registration, [_largest_event()] * BATCH_SIZE) + whole = len(delivery_module._request_body({"events": queued})) + assert whole > MAX_EVENTS_REQUEST_BYTES, whole + sent = _recording_post(monkeypatch) + delivery = CollectorDelivery(spool, registration, development_collector_url=address) + events_path = f"/v1/runs/{run.run_id}/events" + try: + assert delivery.flush_once() + assert delivery.flush_once() + assert not spool.has_pending() + finally: + delivery.close() + spool.close() + requests = [(size, code) for where, size, code in sent if where == events_path] + assert [code for _, code in requests] == [202, 202] + assert all(size <= MAX_EVENTS_REQUEST_BYTES for size, _ in requests), requests + assert _local_only_reason(path, run) is None + assert capsys.readouterr().err == "" + document = reader.get(f"/v1/runs/{run.run_id}") + assert document.status_code == 200, document.text + assert [entry["event_id"] for entry in document.json()["events"]] == [ + event["event_id"] for event in queued + ] + + +def test_the_collector_takes_a_request_body_of_exactly_the_limit( + tmp_path, monkeypatch, live_collector +): + """The byte budget is the collector's limit itself, with no margin: a + request of exactly that many bytes is accepted.""" + address, reader = live_collector + monkeypatch.setenv("HF_TOKEN", "hf_test_member") + run = TelemetryRun(uuid.uuid4().hex, "UK", "contract") + registration = run.as_registration() + spool = EventSpool(tmp_path / "events.sqlite3") + try: + spool.register(registration) + event = spool.append(registration, {"event_type": "run", "status": "started"}) + finally: + spool.close() + token = _collector_token(address, "hf_test_member") + registered = delivery_module._http_post( + address + RUN_REGISTRATION_PATH, registration, token + ) + assert registered[0] == 201, registered + # Pad the one event until its request is the limit to the byte. + spare = MAX_EVENTS_REQUEST_BYTES - len( + delivery_module._request_body({"events": [{**event, "details": {"pad": ""}}]}) + ) + payload = {"events": [{**event, "details": {"pad": "x" * spare}}]} + assert len(delivery_module._request_body(payload)) == MAX_EVENTS_REQUEST_BYTES + + status, body = delivery_module._http_post( + f"{address}/v1/runs/{run.run_id}/events", payload, token + ) + + assert (status, body) == (202, {"accepted": 1, "duplicates": 0}) + document = reader.get(f"/v1/runs/{run.run_id}") + assert document.status_code == 200, document.text + assert [entry["event_id"] for entry in document.json()["events"]] == [ + event["event_id"] + ] From c5760f2bd9235c2a5c85d972f326e2df0de10f9e Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Fri, 9 Oct 2026 21:39:02 -0400 Subject: [PATCH 2/3] Simulate a lost run in the contract test without reaching into the delivery The 404 case wrote a joined-string key into CollectorDelivery._registered, which stops existing once this package keys runs by the pair, as microcosm#1177 now does. It now answers the first registration as accepted without passing it on, so the pinned collector really does not have the run. Co-Authored-By: Claude Opus 5.5 --- tests/test_collector_contract.py | 28 ++++++++++++++++++++++------ 1 file changed, 22 insertions(+), 6 deletions(-) diff --git a/tests/test_collector_contract.py b/tests/test_collector_contract.py index 193e1c4..00c584e 100644 --- a/tests/test_collector_contract.py +++ b/tests/test_collector_contract.py @@ -307,10 +307,24 @@ def test_a_run_the_collector_does_not_have_is_registered_again_and_delivered( spool = EventSpool(path) spool.register(registration) event = spool.append(registration, {"event_type": "run", "status": "started"}) - sent = _recording_post(monkeypatch) + real_post = delivery_module._http_post + sent: list[tuple[str, int]] = [] + + def first_registration_is_lost(url, payload, *args, **kwargs): + where = urlsplit(url).path + if where == RUN_REGISTRATION_PATH and not any( + sent_path == RUN_REGISTRATION_PATH for sent_path, _ in sent + ): + # Answered as registered, but the collector never keeps it: as + # one that lost the run afterwards, its database restored, say. + sent.append((where, 201)) + return 201, {"registered": True} + response = real_post(url, payload, *args, **kwargs) + sent.append((where, response[0])) + return response + + monkeypatch.setattr(delivery_module, "_http_post", first_registration_is_lost) delivery = CollectorDelivery(spool, registration, development_collector_url=address) - # As if the collector no longer had a run this service registered. - delivery._registered.add(delivery_module._registration_key(registration)) try: assert not delivery.flush_once() assert delivery.flush_once() @@ -318,11 +332,13 @@ def test_a_run_the_collector_does_not_have_is_registered_again_and_delivered( finally: delivery.close() spool.close() - assert [(where, code) for where, _, code in sent] == [ + events_path = f"/v1/runs/{run.run_id}/events" + assert sent == [ (TOKEN_EXCHANGE_PATH, 200), - (f"/v1/runs/{run.run_id}/events", 404), (RUN_REGISTRATION_PATH, 201), - (f"/v1/runs/{run.run_id}/events", 202), + (events_path, 404), + (RUN_REGISTRATION_PATH, 201), + (events_path, 202), ] assert _local_only_reason(path, run) is None assert capsys.readouterr().err == "" From 5df012b84e689b6662047febe77b718cee2a4dd8 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Fri, 9 Oct 2026 21:53:34 -0400 Subject: [PATCH 3/3] Check the contract tests' warnings, not all of stderr The collector fixture's Alembic migrations log to stderr, so an empty stderr failed two contract tests whose behaviour had passed against the real collector. They now check that no local-only warning was printed. Co-Authored-By: Claude Opus 5.5 --- tests/test_collector_contract.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_collector_contract.py b/tests/test_collector_contract.py index 00c584e..7bc1e14 100644 --- a/tests/test_collector_contract.py +++ b/tests/test_collector_contract.py @@ -341,7 +341,7 @@ def first_registration_is_lost(url, payload, *args, **kwargs): (events_path, 202), ] assert _local_only_reason(path, run) is None - assert capsys.readouterr().err == "" + assert "local-only" not in capsys.readouterr().err document = reader.get(f"/v1/runs/{run.run_id}") assert document.status_code == 200, document.text assert [entry["event_id"] for entry in document.json()["events"]] == [ @@ -475,7 +475,7 @@ def test_a_full_batch_of_the_largest_events_arrives_in_requests_the_collector_ta assert [code for _, code in requests] == [202, 202] assert all(size <= MAX_EVENTS_REQUEST_BYTES for size, _ in requests), requests assert _local_only_reason(path, run) is None - assert capsys.readouterr().err == "" + assert "local-only" not in capsys.readouterr().err document = reader.get(f"/v1/runs/{run.run_id}") assert document.status_code == 200, document.text assert [entry["event_id"] for entry in document.json()["events"]] == [