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 0000000000..f87bc93c03 --- /dev/null +++ b/changelog.d/telemetry-collector-delivery-followups.fixed.md @@ -0,0 +1 @@ +The telemetry emitter 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. diff --git a/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/collector.py b/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/collector.py index 979d26a078..8743e0d40f 100644 --- a/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/collector.py +++ b/packages/microcosm-build/src/microcosm/build/telemetry_emitter_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/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/constants.py b/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/constants.py index e4331b9039..106ba033b6 100644 --- a/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/constants.py +++ b/packages/microcosm-build/src/microcosm/build/telemetry_emitter_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/packages/microcosm-build/tests/engine_free/shared/test_telemetry_collector_delivery.py b/packages/microcosm-build/tests/engine_free/shared/test_telemetry_collector_delivery.py new file mode 100644 index 0000000000..00375fe93e --- /dev/null +++ b/packages/microcosm-build/tests/engine_free/shared/test_telemetry_collector_delivery.py @@ -0,0 +1,916 @@ +"""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.build.telemetry_emitter import TelemetryRun +from microcosm.build.telemetry_emitter_service import CollectorDelivery, EventSpool +from microcosm.build.telemetry_emitter_service import collector as collector_module +from microcosm.build.telemetry_emitter_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.build.telemetry_emitter_service.database import create_spool_engine +from microcosm.build.telemetry_emitter_service.models import TelemetryRunRecord +from microcosm.build.telemetry_protocol import ( + MAX_TELEMETRY_DETAILS_BYTES, + MAX_TELEMETRY_MESSAGE_CHARS, +) +from microcosm.build.telemetry_sanitization import sanitize_details, sanitize_text + +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/packages/microcosm-build/tests/engine_free/shared/test_telemetry_emitter.py b/packages/microcosm-build/tests/engine_free/shared/test_telemetry_emitter.py index af17b364fa..eb255cfb1a 100644 --- a/packages/microcosm-build/tests/engine_free/shared/test_telemetry_emitter.py +++ b/packages/microcosm-build/tests/engine_free/shared/test_telemetry_emitter.py @@ -432,14 +432,15 @@ def reject(url, payload, token, *, timeout=5.0): @pytest.mark.parametrize( - ("status", "retried"), [(422, False), (404, False), (429, True), (408, True)] + ("status", "retried"), [(422, False), (413, False), (429, True), (408, True)] ) def test_a_settled_collector_rejection_goes_local_only_instead_of_retrying( tmp_path, monkeypatch, capsys, status, retried ) -> 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 or a rate limit still retries.""" + warning. A timeout or a rate limit still retries. A 404 is first answered + by registering the run again: see ``test_telemetry_collector_delivery``.""" spool = EventSpool(tmp_path / "events.sqlite3") registration = _registration() spool.register(registration) diff --git a/packages/microcosm-build/tests/engine_free/shared/test_telemetry_foreign_runs.py b/packages/microcosm-build/tests/engine_free/shared/test_telemetry_foreign_runs.py index 7ba5224d84..01b374b067 100644 --- a/packages/microcosm-build/tests/engine_free/shared/test_telemetry_foreign_runs.py +++ b/packages/microcosm-build/tests/engine_free/shared/test_telemetry_foreign_runs.py @@ -7,6 +7,8 @@ exchange refuses a non-member with 403 and a bad token with 401; the first Hugging Face user to register a run owns it and anyone else gets 403; an unknown run's events get 404, and a batch the collector cannot accept 422. +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 @@ -14,7 +16,9 @@ - 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. + run goes local-only only after a collector answer about the run itself: a + registration 409, a 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 @@ -25,7 +29,9 @@ goes local-only, even from a service that listed the run before. - 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. + the run under an earlier login, unless the collector refuses the run itself + (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 @@ -65,6 +71,7 @@ from microcosm.build.telemetry_emitter_service import leases as leases_module from microcosm.build.telemetry_emitter_service import main as main_module from microcosm.build.telemetry_emitter_service.constants import ( + BATCH_SIZE, DEFAULT_TOKEN_LIFETIME_SECONDS, LOCAL_ONLY_MISSING_CREDENTIAL, LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION, @@ -120,8 +127,11 @@ def __init__( self, rejected_run_ids: set[str] | None = None, conflicted_run_ids: set[str] | None = None, + unfindable_run_ids: set[str] | None = None, ) -> None: 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 ()) self.owners: dict[str, str] = {} # A conflicted run id was registered before with other metadata, so # whoever owns it gets 409 for this registration. @@ -182,7 +192,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, {}) @@ -195,6 +205,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.""" @@ -345,11 +362,27 @@ 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.""" assert host.intrusions == [] - rejected = host.collector.rejected_run_ids conflicted = host.collector.conflicted_run_ids for caller, run, reason, credential, at in host.local_only: # Local-only is final: nothing about the run reaches the collector after. @@ -365,8 +398,7 @@ def _assert_own_credential_only(host: Host) -> None: assert latest == ("register", run, 409), (caller, run, latest) else: assert reason == LOCAL_ONLY_REJECTED_EVENTS, (caller, run, reason) - assert run[0] in rejected, (caller, run) - assert latest == ("events", run, 422), (caller, run, latest) + _assert_events_were_refused(host, caller, run, at) continue if reason == LOCAL_ONLY_MISSING_CREDENTIAL: assert credential is None @@ -384,10 +416,7 @@ def _assert_own_credential_only(host: Host) -> None: assert latest == ("events", run, 403), (run, latest) else: assert reason == LOCAL_ONLY_REJECTED_EVENTS - 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 @@ -788,7 +817,9 @@ def hold(): lease.release() -@pytest.mark.parametrize("answer", ["unacceptable_events", "registration_conflict"]) +@pytest.mark.parametrize( + "answer", ["unacceptable_events", "unfindable_run", "registration_conflict"] +) def test_an_orphan_the_collector_refuses_goes_local_only_without_a_warning_here( host, capsys, answer ): @@ -798,6 +829,9 @@ def test_an_orphan_the_collector_refuses_goes_local_only_without_a_warning_here( if answer == "unacceptable_events": host.collector.rejected_run_ids.add(build_a.run[0]) reason = LOCAL_ONLY_REJECTED_EVENTS + elif answer == "unfindable_run": + host.collector.unfindable_run_ids.add(build_a.run[0]) + reason = LOCAL_ONLY_REJECTED_EVENTS else: # The run id is already Alice's with other metadata: a 409 reaches # only the owner, and every owner would get it. @@ -808,11 +842,74 @@ def test_an_orphan_the_collector_refuses_goes_local_only_without_a_warning_here( host.emit(build_b) host.flush(build_b) + if answer == "unfindable_run": + # The first 404 has the run registered again; the second settles it. + assert host.states()[build_a.run] == ("pending", None) + 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) == [404, 404] assert host.states()[build_a.run] == ("local_only", reason) assert host.collector.accepted[build_b.run] == set(build_b.appended) # The warning says "this run"; another build's run is not this build's. assert "local-only" not in capsys.readouterr().err + _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) + + 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) def test_a_shared_run_id_refuses_the_second_user_its_own_run_only(host): @@ -1164,9 +1261,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}") @@ -1184,6 +1286,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()), @@ -1212,6 +1316,7 @@ def _start_build(host: Host, build: tuple[str | None, int]) -> Producer: operations=OPERATIONS, rejected=st.sets(st.integers(0, 7), max_size=2), conflicted=st.sets(st.integers(0, 7), max_size=2), + 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)), ) @@ -1223,6 +1328,7 @@ def _start_build(host: Host, build: tuple[str | None, int]) -> Producer: operations=[("interleave", 0, 1, True)], rejected=set(), conflicted=set(), + unfindable=set(), rescue_logins=(MEMBER + "alice", MEMBER + "bob"), ) # From review: a service refused for every pending run, whose own run is not @@ -1232,16 +1338,35 @@ def _start_build(host: Host, build: tuple[str | None, int]) -> Producer: operations=[], rejected=set(), conflicted=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(), + 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, rescue_logins + tmp_path_factory, + builds, + operations, + rejected, + conflicted, + unfindable, + rescue_logins, ): host = Host( tmp_path_factory.mktemp("foreign"), FakeCollector( {f"run-{index}" for index in rejected}, {f"run-{index}" for index in conflicted}, + {f"run-{index}" for index in unfindable}, ), ) try: @@ -1253,6 +1378,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 @@ -1314,7 +1443,12 @@ def other_flushes(other=other, exits=exits) -> None: LOCAL_ONLY_REJECTED_REGISTRATION, ), producer.run continue - if run_id in host.collector.rejected_run_ids and producer.appended: + refused = ( + host.collector.rejected_run_ids | host.collector.unfindable_run_ids + ) + if run_id in refused and producer.appended: + # Its owner, whoever that is, gets 422, or 404 twice: a + # refusal of the batch. if host.queued(producer.run): assert after[producer.run] == ( "local_only",