From 7d42a030d3b104804ed64c8c12b0f651b44ddaa5 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Fri, 9 Oct 2026 19:58:58 -0400 Subject: [PATCH 1/2] Make a settled collector rejection local-only instead of retrying it forever Ports PolicyEngine/microcosm 0e80b0361 (microcosm#1147) into the emitter's copy of the telemetry service, on top of the foreign-run fix (#5). An events answer other than 202, 401 or 403, and a registration answer other than 201, 401, 403 or 409, was retried without end. A collector that answers a batch with 422 (an event shape it does not accept), 404, 413 or 400 therefore wedged that run's queue for good, and the retry delay it set gates every later delivery pass of the service. A 4xx other than a timeout (408) or a rate limit (429) on a run's events now makes the run local-only (`collector_rejected_events`) with one warning that names the status; the same rule covers a rejected registration (`run_registration_rejected`). Both are about the run and not the caller, so a service applies them to a run it has adopted too, without a warning in its own build. 401 still exchanges the token again, 403 is still about the login, and 408, 429 and 5xx still retry. collector.py now equals microcosm#1177's (head 615d84f89, which contains 0e80b0361) apart from the package path. Tests: Microcosm's delivery test, plus the registration side, every 4xx on both requests, a 401 re-exchange, and a contract case against the pinned collector (422 for an event outside its schema, 404 for an unknown run). The foreign-run tests now expect an adopted run refused with 422, at registration or at its events, to go local-only, accept that only when the adopting service's latest answer is that 422 for that run, and check that a refused orphan sets no retry delay. Co-Authored-By: Claude Opus 5.5 --- README.md | 2 +- ...etry-settled-rejection-local-only.fixed.md | 1 + .../telemetry/service/collector.py | 34 ++- .../telemetry/service/constants.py | 6 + tests/telemetry/test_delivery.py | 286 ++++++++++++++++++ tests/telemetry/test_foreign_runs.py | 147 +++++++-- tests/test_collector_contract.py | 64 ++++ 7 files changed, 502 insertions(+), 38 deletions(-) create mode 100644 changelog.d/telemetry-settled-rejection-local-only.fixed.md diff --git a/README.md b/README.md index c6fa4ac6..b1ed40cb 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 retain events for retry. 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 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. diff --git a/changelog.d/telemetry-settled-rejection-local-only.fixed.md b/changelog.d/telemetry-settled-rejection-local-only.fixed.md new file mode 100644 index 00000000..924d5e5b --- /dev/null +++ b/changelog.d/telemetry-settled-rejection-local-only.fixed.md @@ -0,0 +1 @@ +The telemetry service no longer retries a settled collector rejection forever. A 4xx other than a timeout (408) or a rate limit (429) on a run's events or registration (an event shape the collector does not accept, say) now makes that run local-only with a recorded reason, `collector_rejected_events` for a refused batch, and one warning that names the status. Before, the service sent the same request again without end, and the retry delay it set held back every other run it delivers. The rule is about the run and not about who sends it, so a service that has taken over another build's run applies it there too, with no warning in its own build. An expired collector token (401) is still exchanged again, a refused login (403) is handled as before, and a timeout, a rate limit or a server error is still retried. This ports PolicyEngine/microcosm#1147's rule, 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 d42ea84c..2c333f28 100644 --- a/src/microcosm_emitter/telemetry/service/collector.py +++ b/src/microcosm_emitter/telemetry/service/collector.py @@ -31,6 +31,7 @@ LOCAL_ONLY_MISSING_CREDENTIAL, LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION, LOCAL_ONLY_REJECTED_CREDENTIAL, + LOCAL_ONLY_REJECTED_EVENTS, LOCAL_ONLY_REJECTED_REGISTRATION, LOOPBACK_HOSTS, MAX_HTTP_RESPONSE_BYTES, @@ -41,6 +42,7 @@ OWN_LEASE_TIMEOUT_SECONDS, PRODUCTION_COLLECTOR_URL, REJECTED_CREDENTIAL_MESSAGE, + REJECTED_EVENTS_MESSAGE, RUN_EVENTS_PATH_TEMPLATE, RUN_REGISTRATION_PATH, TOKEN_EXCHANGE_PATH, @@ -134,6 +136,18 @@ def _huggingface_token() -> str | None: ) +#: Client errors worth retrying: a timeout and a rate limit pass with time. +#: Every other 4xx is the collector's settled answer to the same request, so +#: retrying it only wedges the queue (a payload shape it does not accept, say). +_RETRYABLE_CLIENT_ERRORS = frozenset( + {HTTPStatus.REQUEST_TIMEOUT, HTTPStatus.TOO_MANY_REQUESTS} +) + + +def _permanent_client_error(status: int) -> bool: + return 400 <= int(status) < 500 and status not in _RETRYABLE_CLIENT_ERRORS + + def _run(registration: Mapping[str, Any]) -> tuple[str, str]: """Identify a producer's run. Ids may contain any separator, so the pair itself is the key: joined into one string, two runs could share it.""" @@ -358,6 +372,13 @@ def _deliver(self, registration: Mapping[str, Any]) -> bool: LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION, caller_dependent=True, ) + elif _permanent_client_error(status): + self._make_local_only( + registration, + LOCAL_ONLY_REJECTED_EVENTS, + caller_dependent=False, + message=REJECTED_EVENTS_MESSAGE.format(status=int(status)), + ) else: self._defer_retry() return False @@ -440,9 +461,9 @@ def _register(self, registration: Mapping[str, Any], token: str) -> bool: LOCAL_ONLY_REJECTED_REGISTRATION, caller_dependent=True, ) - elif status == HTTPStatus.CONFLICT: + elif status == HTTPStatus.CONFLICT or _permanent_client_error(status): # A conflict reaches only the run's owner, so it is about the - # registration itself. + # registration itself, as is any other settled 4xx. self._make_local_only( registration, LOCAL_ONLY_REJECTED_REGISTRATION, @@ -458,6 +479,7 @@ def _make_local_only( reason: str, *, caller_dependent: bool, + message: str = REJECTED_CREDENTIAL_MESSAGE, ) -> None: """Stop uploading a run, unless the reason is only this service's. @@ -475,17 +497,17 @@ def _make_local_only( self._passed_over[run] = self._credential return self.spool.make_local_only(*run, reason) - # The warning says "this run": another build's run is not this build's + # 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: - self._warn_denied(run) + self._warn_denied(run, message) def _defer_retry(self) -> None: self._next_attempt_at = time.monotonic() + self._retry_seconds self._retry_seconds = min(MAX_RETRY_SECONDS, self._retry_seconds * 2) - def _warn_denied(self, run: tuple[str, str]) -> None: + def _warn_denied(self, run: tuple[str, str], message: str) -> None: if run in self._warned_denied: return - print(REJECTED_CREDENTIAL_MESSAGE, file=sys.stderr, flush=True) + print(message, file=sys.stderr, flush=True) self._warned_denied.add(run) diff --git a/src/microcosm_emitter/telemetry/service/constants.py b/src/microcosm_emitter/telemetry/service/constants.py index 04945d01..3c447de0 100644 --- a/src/microcosm_emitter/telemetry/service/constants.py +++ b/src/microcosm_emitter/telemetry/service/constants.py @@ -25,6 +25,7 @@ LOCAL_ONLY_REJECTED_CREDENTIAL: Final = "huggingface_credential_rejected" LOCAL_ONLY_REJECTED_REGISTRATION: Final = "run_registration_rejected" LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION: Final = "collector_authorization_rejected" +LOCAL_ONLY_REJECTED_EVENTS: Final = "collector_rejected_events" LOCAL_ONLY_PRE_ELIGIBILITY: Final = "created_before_upload_eligibility" NO_CREDENTIAL_MESSAGE: Final = ( @@ -36,6 +37,11 @@ "credential was not accepted as a PolicyEngine organization member. The " "dataset build will continue." ) +REJECTED_EVENTS_MESSAGE: Final = ( + "Microcosm telemetry is local-only for this run: the collector rejected its " + "events (HTTP {status}), so they will not be retried. 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_delivery.py b/tests/telemetry/test_delivery.py index b36f2f40..b4d2b071 100644 --- a/tests/telemetry/test_delivery.py +++ b/tests/telemetry/test_delivery.py @@ -2,9 +2,11 @@ import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from types import SimpleNamespace +from urllib.parse import urlsplit import pytest from alembic import command +from sqlalchemy.orm import Session from microcosm_emitter.telemetry.client import ( TelemetryRun, @@ -17,7 +19,13 @@ from microcosm_emitter.telemetry.service import spool as spool_module from microcosm_emitter.telemetry.service.collector import CollectorDelivery from microcosm_emitter.telemetry.service.constants import ( + LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION, + LOCAL_ONLY_REJECTED_EVENTS, + LOCAL_ONLY_REJECTED_REGISTRATION, + MAX_RETRY_SECONDS, PRODUCTION_COLLECTOR_URL, + RUN_REGISTRATION_PATH, + TOKEN_EXCHANGE_PATH, ) from microcosm_emitter.telemetry.service.database import ( create_spool_engine, @@ -28,6 +36,7 @@ migration_head_revision, ) from microcosm_emitter.telemetry.service.models import ( + TelemetryRunRecord, serialized_json_length, ) from microcosm_emitter.telemetry.service.sanitization import ( @@ -63,6 +72,19 @@ def make(*args, **kwargs) -> CollectorDelivery: delivery.close() +def _local_only_reason(spool_path, registration) -> str | None: + engine = create_spool_engine(spool_path) + try: + with Session(engine) as session: + run = session.get( + TelemetryRunRecord, + (registration["run_id"], registration["producer_id"]), + ) + return run.local_only_reason + finally: + engine.dispose() + + def _event(stage_id: str = "compile_targets") -> dict[str, object]: return { "timestamp": "2026-10-02T10:00:00+00:00", @@ -291,6 +313,270 @@ def reject(url, payload, token, *, timeout=5.0): assert capsys.readouterr().err.count("local-only for this run") == 1 +@pytest.mark.parametrize( + ("status", "retried"), + [ + (422, False), + (404, False), + (413, False), + (400, False), + (429, True), + (408, True), + (503, True), + ], +) +def test_a_settled_collector_rejection_goes_local_only_instead_of_retrying( + tmp_path, monkeypatch, capsys, make_delivery, 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, a rate limit or a server error still retries.""" + spool_path = tmp_path / "events.sqlite3" + spool = EventSpool(spool_path) + registration = _registration() + spool.register(registration) + spool.append(registration, _event()) + monkeypatch.setattr(collector_module, "_huggingface_token", lambda: "hf-member") + paths: list[str] = [] + clock = [0.0] + monkeypatch.setattr( + collector_module, "time", SimpleNamespace(monotonic=lambda: clock[0]) + ) + + def fake_post(url, payload, token, *, timeout=5.0): + paths.append(urlsplit(url).path) + if url.endswith(TOKEN_EXCHANGE_PATH): + return 200, {"access_token": "collector-token", "expires_in": 3600} + if url.endswith(RUN_REGISTRATION_PATH): + return 201, {"registered": True} + return status, {"detail": "rejected"} + + monkeypatch.setattr(collector_module, "_http_post", fake_post) + delivery = make_delivery( + spool, registration, development_collector_url="http://127.0.0.1:8080" + ) + events_path = "/v1/runs/run-a/events" + assert not delivery.flush_once() + assert spool.has_pending() + assert paths.count(events_path) == 1 + if retried: + assert spool.pending_runs() == [registration] + assert _local_only_reason(spool_path, registration) is None + assert "local-only" not in capsys.readouterr().err + # Once the retry delay has passed, the same batch is offered again. + clock[0] += MAX_RETRY_SECONDS + 1 + assert not delivery.flush_once() + assert paths.count(events_path) == 2 + assert spool.pending_runs() == [registration] + return + assert spool.pending_runs() == [] + assert _local_only_reason(spool_path, registration) == LOCAL_ONLY_REJECTED_EVENTS + clock[0] += MAX_RETRY_SECONDS + 1 + assert not delivery.flush_once() + assert paths.count(events_path) == 1 + err = capsys.readouterr().err + assert err.count("local-only for this run") == 1 + assert f"HTTP {status}" in err + + +@pytest.mark.parametrize( + ("status", "retried"), + [ + (409, False), + (422, False), + (404, False), + (413, False), + (400, False), + (429, True), + (408, True), + (503, True), + ], +) +def test_a_settled_registration_rejection_goes_local_only_instead_of_retrying( + tmp_path, monkeypatch, capsys, make_delivery, status, retried +) -> None: + """The same rule covers a run's registration: a settled 4xx makes the run + local-only before any of its events are sent; a timeout, a rate limit or a + server error is retried.""" + spool_path = tmp_path / "events.sqlite3" + spool = EventSpool(spool_path) + registration = _registration() + spool.register(registration) + spool.append(registration, _event()) + monkeypatch.setattr(collector_module, "_huggingface_token", lambda: "hf-member") + paths: list[str] = [] + clock = [0.0] + monkeypatch.setattr( + collector_module, "time", SimpleNamespace(monotonic=lambda: clock[0]) + ) + + def fake_post(url, payload, token, *, timeout=5.0): + paths.append(urlsplit(url).path) + if url.endswith(TOKEN_EXCHANGE_PATH): + return 200, {"access_token": "collector-token", "expires_in": 3600} + if url.endswith(RUN_REGISTRATION_PATH): + return status, {"detail": "rejected"} + raise AssertionError("events sent for a run that was not registered") + + monkeypatch.setattr(collector_module, "_http_post", fake_post) + delivery = make_delivery( + spool, registration, development_collector_url="http://127.0.0.1:8080" + ) + assert not delivery.flush_once() + assert paths.count(RUN_REGISTRATION_PATH) == 1 + clock[0] += MAX_RETRY_SECONDS + 1 + assert not delivery.flush_once() + assert spool.has_pending() + if retried: + assert paths.count(RUN_REGISTRATION_PATH) == 2 + assert spool.pending_runs() == [registration] + assert _local_only_reason(spool_path, registration) is None + assert "local-only" not in capsys.readouterr().err + return + assert paths.count(RUN_REGISTRATION_PATH) == 1 + assert spool.pending_runs() == [] + assert ( + _local_only_reason(spool_path, registration) == LOCAL_ONLY_REJECTED_REGISTRATION + ) + assert capsys.readouterr().err.count("local-only for this run") == 1 + + +def test_only_a_settled_client_error_is_permanent() -> None: + """Over every status: permanent exactly for a 4xx other than a timeout and + a rate limit. 401 and 403 count here too; the delivery answers them first.""" + for status in range(100, 600): + expected = status // 100 == 4 and status not in (408, 429) + assert collector_module._permanent_client_error(status) is expected, status + + +@pytest.mark.parametrize("refused", ["registration", "events"]) +def test_every_client_error_has_one_outcome( + tmp_path, monkeypatch, capsys, refused +) -> 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.""" + spool_path = tmp_path / "events.sqlite3" + spool = EventSpool(spool_path) + monkeypatch.setattr(collector_module, "_huggingface_token", lambda: "hf-member") + clock = [0.0] + monkeypatch.setattr( + collector_module, "time", SimpleNamespace(monotonic=lambda: clock[0]) + ) + settled_reason = ( + LOCAL_ONLY_REJECTED_REGISTRATION + if refused == "registration" + else LOCAL_ONLY_REJECTED_EVENTS + ) + login_reason = ( + LOCAL_ONLY_REJECTED_REGISTRATION + if refused == "registration" + else LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION + ) + for status in range(400, 500): + registration = _registration(f"run-{status}") + spool.register(registration) + queued = spool.append(registration, _event()) + refused_path = ( + RUN_REGISTRATION_PATH + if refused == "registration" + else f"/v1/runs/run-{status}/events" + ) + sent: list[str] = [] + + def fake_post(url, payload, token, *, timeout=5.0, status=status, sent=sent): + path = urlsplit(url).path + sent.append(path) + if path == TOKEN_EXCHANGE_PATH: + return 200, {"access_token": "collector-token", "expires_in": 3600} + if path == RUN_REGISTRATION_PATH and refused == "events": + return 201, {"registered": True} + return status, {"detail": "refused"} + + monkeypatch.setattr(collector_module, "_http_post", fake_post) + delivery = CollectorDelivery( + spool, registration, development_collector_url="http://127.0.0.1:8080" + ) + try: + assert not delivery.flush_once(), status + assert sent.count(refused_path) == 1, status + # A second pass with no time passed: only a retry delay or a + # local-only run keeps it from asking again. + first = len(sent) + assert not delivery.flush_once(), status + again = sent[first:] + reason = _local_only_reason(spool_path, registration) + warnings = capsys.readouterr().err + if status == 401: + assert again[0] == TOKEN_EXCHANGE_PATH, status + assert reason is None, status + assert warnings == "", status + continue + assert again == [], (status, again) + if status in (408, 429): + assert reason is None, status + assert warnings == "", status + clock[0] += MAX_RETRY_SECONDS + 1 + assert not delivery.flush_once(), status + assert sent[first:] == [refused_path], status + 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 + # Nothing is sent again, however long the service waits. + clock[0] += MAX_RETRY_SECONDS + 1 + assert not delivery.flush_once(), status + assert sent[first:] == [], status + finally: + delivery.close() + # Leave nothing pending for the next status's service to adopt. + spool.acknowledge([queued["event_id"]]) + + +@pytest.mark.parametrize("refused", [RUN_REGISTRATION_PATH, "/v1/runs/run-a/events"]) +def test_an_expired_collector_token_is_exchanged_again_not_made_local_only( + tmp_path, monkeypatch, capsys, make_delivery, refused +) -> None: + """A 401 is a 4xx too, but it is about the short-lived collector token, + which a new exchange replaces.""" + spool_path = tmp_path / "events.sqlite3" + spool = EventSpool(spool_path) + registration = _registration() + spool.register(registration) + spool.append(registration, _event()) + monkeypatch.setattr(collector_module, "_huggingface_token", lambda: "hf-member") + requests: list[tuple[str, str]] = [] + + def fake_post(url, payload, token, *, timeout=5.0): + path = urlsplit(url).path + requests.append((path, token)) + if path == TOKEN_EXCHANGE_PATH: + exchanges = sum(1 for sent, _ in requests if sent == TOKEN_EXCHANGE_PATH) + return 200, {"access_token": f"collector-{exchanges}", "expires_in": 3600} + if path == refused and token == "collector-1": + return 401, {"detail": "Collector credential is invalid."} + if path == RUN_REGISTRATION_PATH: + return 201, {"registered": True} + return 202, {"accepted": 1, "duplicates": 0} + + monkeypatch.setattr(collector_module, "_http_post", fake_post) + delivery = make_delivery( + spool, registration, development_collector_url="http://127.0.0.1:8080" + ) + + assert not delivery.flush_once() + assert delivery.flush_once() + + assert [token for path, token in requests if path == refused] == [ + "collector-1", + "collector-2", + ] + assert not spool.has_pending() + assert _local_only_reason(spool_path, registration) is None + assert "local-only" not in capsys.readouterr().err + + def test_identity_provider_outage_keeps_events_eligible_for_retry( tmp_path, monkeypatch, make_delivery ) -> None: diff --git a/tests/telemetry/test_foreign_runs.py b/tests/telemetry/test_foreign_runs.py index ec68b7fa..eabd0795 100644 --- a/tests/telemetry/test_foreign_runs.py +++ b/tests/telemetry/test_foreign_runs.py @@ -6,8 +6,9 @@ (PolicyEngine/calibration-diagnostics, ``telemetry-service``): the token 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; the -owner registering it again with other metadata gets 409; an unknown run's -events get 404, and a batch the collector cannot accept 422. +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. 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 @@ -15,7 +16,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. + run goes local-only only after a collector answer about the run itself: a + registration 409, or a registration or batch refused with 422. - 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 @@ -26,11 +28,8 @@ 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. - -This service retries a batch the collector refuses with 422 rather than making -its run local-only, so such a run stays queued for retry; the tests check only -that no service's credential decides it. + the run under an earlier login, unless the collector refuses the run itself + (409 or 422), which makes it local-only. """ from __future__ import annotations @@ -71,6 +70,7 @@ LOCAL_ONLY_MISSING_CREDENTIAL, LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION, LOCAL_ONLY_REJECTED_CREDENTIAL, + LOCAL_ONLY_REJECTED_EVENTS, LOCAL_ONLY_REJECTED_REGISTRATION, MAX_RETRY_SECONDS, ORPHAN_IDLE_SECONDS, @@ -123,8 +123,13 @@ def __init__( self, rejected_run_ids: set[str] | None = None, conflicted_run_ids: set[str] | None = None, + malformed_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 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 ()) self.owners: dict[str, str] = {} # A conflicted run id was registered before with other metadata, so # whoever owns it gets 409 for this registration. @@ -175,6 +180,8 @@ def post(self, caller: Run, url: str, payload, token: str) -> tuple[int, dict]: user = token.removeprefix("session:") if path == RUN_REGISTRATION_PATH: run = (payload["run_id"], payload["producer_id"]) + if run[0] in self.malformed_run_ids: + return self._answer(caller, "register", run, 422, {}) if self.owners.setdefault(run[0], user) != user: return self._answer(caller, "register", run, 403, {}) metadata = {k: v for k, v in payload.items() if k != "producer_id"} @@ -289,7 +296,10 @@ def emit(self, producer: Producer, count: int = 1) -> None: event = producer.spool.append(producer.registration, EVENT) producer.appended.append(event["event_id"]) - def flush(self, producer: Producer) -> bool: + def flush(self, producer: Producer, *, advance: bool = True) -> bool: + """Run one delivery pass; ``advance=False`` runs it with no time passed + since the last one, so a retry delay the service set still holds.""" + assert producer.alive def post(url, payload, token, *, timeout=5.0): @@ -300,7 +310,8 @@ def post(url, payload, token, *, timeout=5.0): self.intrusions.append((producer.run, url, about)) return status, body - self.clock += MAX_RETRY_SECONDS + 1 + if advance: + self.clock += MAX_RETRY_SECONDS + 1 with ( mock.patch.object( collector_module, "_huggingface_token", lambda: producer.credential @@ -376,9 +387,16 @@ def _assert_own_credential_only(host: Host) -> None: if caller != run: # 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) + continue assert reason == LOCAL_ONLY_REJECTED_REGISTRATION, (caller, run, reason) - assert run[0] in conflicted - assert latest == ("register", run, 409), (caller, run, latest) + if run[0] in host.collector.malformed_run_ids: + assert latest == ("register", run, 422), (caller, run, latest) + else: + assert run[0] in conflicted, (caller, run) + assert latest == ("register", run, 409), (caller, run, latest) continue if reason == LOCAL_ONLY_MISSING_CREDENTIAL: assert credential is None @@ -388,13 +406,19 @@ def _assert_own_credential_only(host: Host) -> None: first_refused = host.collector.refused_at.get((caller, credential), at) assert first_refused < at, (caller, credential) elif reason == LOCAL_ONLY_REJECTED_REGISTRATION: - assert latest in (("register", run, 403), ("register", run, 409)), ( + assert latest in ( + ("register", run, 403), + ("register", run, 409), + ("register", run, 422), + ), (run, latest) + elif reason == LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION: + 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, ) - else: - assert reason == LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION, reason - assert latest == ("events", run, 403), (run, latest) # A refused credential is exchanged once per service, not on every pass. refused = host.collector.refused assert len(set(refused)) == len(refused), refused @@ -736,32 +760,82 @@ def hold(): lease.release() +# What the collector answers about a run itself, whoever asks: the reason the +# run then keeps, and every status sent about it. +RUN_REFUSALS = { + "unacceptable_events": (LOCAL_ONLY_REJECTED_EVENTS, [201, 422]), + "unacceptable_registration": (LOCAL_ONLY_REJECTED_REGISTRATION, [422]), + "registration_conflict": (LOCAL_ONLY_REJECTED_REGISTRATION, [409]), +} + + +def _refuse_run(host: Host, run: Run, answer: str) -> None: + if answer == "unacceptable_events": + host.collector.rejected_run_ids.add(run[0]) + elif answer == "unacceptable_registration": + host.collector.malformed_run_ids.add(run[0]) + else: + # The run id is already Alice's with other metadata. + host.collector.owners[run[0]] = "alice" + host.collector.metadata[run[0]] = {"pipeline": "something-else"} + + +@pytest.mark.parametrize("answer", RUN_REFUSALS) def test_an_orphan_the_collector_refuses_goes_local_only_without_a_warning_here( - host, capsys + host, capsys, answer ): - """A 409 reaches only the run's owner, and every owner would get it.""" + """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.""" + reason, statuses = RUN_REFUSALS[answer] build_a = host.start(MEMBER + "alice") - host.emit(build_a) + host.emit(build_a, 2) host.kill(build_a) - # The run id is already Alice's with other metadata. - host.collector.owners[build_a.run[0]] = "alice" - host.collector.metadata[build_a.run[0]] = {"pipeline": "something-else"} + _refuse_run(host, build_a.run, answer) build_b = host.start(MEMBER + "alice") host.emit(build_b) + host.flush(build_b) host.flush(build_b) - assert host.states()[build_a.run] == ( - "local_only", - LOCAL_ONLY_REJECTED_REGISTRATION, - ) + assert host.states()[build_a.run] == ("local_only", reason) + # One refusal settles it: the run is neither retried nor left queued for + # another service. + assert [ + status + for _, _, about, status in host.collector.requests + if about == build_a.run + ] == statuses + assert host.queued(build_a.run) == 2 + assert build_a.run not in host.collector.accepted 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) +@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.""" + + build_a = host.start(MEMBER + "alice") + host.emit(build_a) + host.kill(build_a) + _refuse_run(host, build_a.run, answer) + build_b = host.start(MEMBER + "alice") + host.emit(build_b) + host.flush(build_b) + assert host.states()[build_a.run][0] == "local_only" + + host.emit(build_b) + host.flush(build_b, advance=False) + + assert host.collector.accepted[build_b.run] == set(build_b.appended) + _assert_own_credential_only(host) + + def test_a_shared_run_id_refuses_the_second_user_its_own_run_only(host): build_a = host.start(MEMBER + "alice", run_id="shared") build_b = host.start(MEMBER + "bob", run_id="shared") @@ -1293,6 +1367,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), + malformed=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)), ) @@ -1304,6 +1379,7 @@ def _start_build(host: Host, build: tuple[str | None, int]) -> Producer: operations=[("interleave", 0, 1, True)], rejected=set(), conflicted=set(), + malformed=set(), rescue_logins=(MEMBER + "alice", MEMBER + "bob"), ) # From review: a service refused for every pending run, whose own run is not @@ -1313,16 +1389,18 @@ def _start_build(host: Host, build: tuple[str | None, int]) -> Producer: operations=[], rejected=set(), conflicted=set(), + malformed=set(), rescue_logins=(OUTSIDER, EXPIRED), ) 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, malformed, 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 malformed}, ), ) try: @@ -1388,16 +1466,23 @@ def other_flushes(other=other, exits=exits) -> None: run_id = producer.run[0] if not host.queued(producer.run): assert after[producer.run] == ("pending", None), producer.run - elif run_id in host.collector.conflicted_run_ids: - # Its owner, whoever that is, gets 409: a refusal about the run. + elif ( + run_id in host.collector.malformed_run_ids + or run_id in host.collector.conflicted_run_ids + ): + # Whoever registers it gets 422, or its owner gets 409: a + # refusal about the run. assert after[producer.run] == ( "local_only", LOCAL_ONLY_REJECTED_REGISTRATION, ), producer.run continue elif run_id in host.collector.rejected_run_ids: - # Retried, not made local-only: still pending for its owner. - assert after[producer.run] == ("pending", None), producer.run + # Its owner, whoever that is, gets 422: a refusal of the batch. + assert after[producer.run] == ( + "local_only", + LOCAL_ONLY_REJECTED_EVENTS, + ), producer.run continue assert after[producer.run] == ("pending", None), producer.run assert host.queued(producer.run) == 0, producer.run diff --git a/tests/test_collector_contract.py b/tests/test_collector_contract.py index 3a0be616..51afe679 100644 --- a/tests/test_collector_contract.py +++ b/tests/test_collector_contract.py @@ -5,13 +5,18 @@ import threading import time import uuid +from urllib.parse import urlsplit import pytest from sqlalchemy.engine import make_url +from sqlalchemy.orm import Session from microcosm_emitter.telemetry.client import LocalTelemetryEmitter, TelemetryRun 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.database import create_spool_engine +from microcosm_emitter.telemetry.service.models import TelemetryRunRecord from microcosm_emitter.telemetry.service.spool import EventSpool pytestmark = pytest.mark.collector @@ -197,3 +202,62 @@ def lose_ack(url, *args, **kwargs): assert [entry["event_id"] for entry in response.json()["events"]] == [ event["event_id"] ] + + +@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 +): + """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.""" + 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_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) + 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 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() + warnings = capsys.readouterr().err + assert warnings.count("local-only for this run") == 1 + assert f"HTTP {status}" 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 == 404 + else: + assert document.status_code == 200, document.text + assert document.json()["events"] == [] From 9d4ee06778a39ccb7f4cfdb4438cb5c2127dc963 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Fri, 9 Oct 2026 21:37:03 -0400 Subject: [PATCH 2/2] Say that a refused registration keeps the warning a 409 already had Review round 1 on #6: the changelog promised a warning naming the status for a refused registration too, but `_register` passes no message, so the service prints the existing rejected-credential text for any settled registration refusal, as it already did for 409. Only a refused batch's warning names the status. Co-Authored-By: Claude Opus 5.5 --- changelog.d/telemetry-settled-rejection-local-only.fixed.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/changelog.d/telemetry-settled-rejection-local-only.fixed.md b/changelog.d/telemetry-settled-rejection-local-only.fixed.md index 924d5e5b..5ddf10ba 100644 --- a/changelog.d/telemetry-settled-rejection-local-only.fixed.md +++ b/changelog.d/telemetry-settled-rejection-local-only.fixed.md @@ -1 +1 @@ -The telemetry service no longer retries a settled collector rejection forever. A 4xx other than a timeout (408) or a rate limit (429) on a run's events or registration (an event shape the collector does not accept, say) now makes that run local-only with a recorded reason, `collector_rejected_events` for a refused batch, and one warning that names the status. Before, the service sent the same request again without end, and the retry delay it set held back every other run it delivers. The rule is about the run and not about who sends it, so a service that has taken over another build's run applies it there too, with no warning in its own build. An expired collector token (401) is still exchanged again, a refused login (403) is handled as before, and a timeout, a rate limit or a server error is still retried. This ports PolicyEngine/microcosm#1147's rule, so both packages' services treat a shared queue's runs alike. +The telemetry service no longer retries a settled collector rejection forever. A 4xx other than a timeout (408) or a rate limit (429) on a run's events or registration (an event shape the collector does not accept, say) now makes that run local-only with a recorded reason and one warning. For a refused batch the reason is `collector_rejected_events` and the warning names the status; a refused registration keeps the reason and the warning a 409 already had. Before, the service sent the same request again without end, and the retry delay it set held back every other run it delivers. The rule is about the run and not about who sends it, so a service that has taken over another build's run applies it there too, with no warning in its own build. An expired collector token (401) is still exchanged again, a refused login (403) is handled as before, and a timeout, a rate limit or a server error is still retried. This ports PolicyEngine/microcosm#1147's rule, so both packages' services treat a shared queue's runs alike.