diff --git a/README.md b/README.md index fc94a0db..66bee0c1 100644 --- a/README.md +++ b/README.md @@ -40,7 +40,7 @@ The adapter bounds socket calls and reports telemetry failures without failing t `block()` finishes the current stage and records that validation checks refused the build, including the phase, check identifiers, failure count and optional check statuses. `fail(error_code=...)` preserves the caller's failure code alongside the exception type and failure class. Microcosm classifies interruptions, process termination and memory failures; this package forwards those classifications rather than duplicating them. Both outcomes finish telemetry, so later heartbeat or parent-exit handling cannot replace them. -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. Missing or rejected credentials permanently mark that producer's events as local-only. The emitter sends a credential only when it is visible ASCII (letters, digits and punctuation), as every Hugging Face credential and collector token is. It refuses any other value, such as one with a pasted space, a tab, a line break or a non-ASCII character, before sending it: a bearer token (RFC 6750) cannot contain those characters, so the collector would refuse it, and Python's HTTP client raises an error that quotes the whole header value for a line break. Such a credential counts as rejected: the telemetry module marks the producer's events local-only with a warning that the credential contains characters that cannot be sent, collector submissions end `failed` with `CREDENTIAL_REJECTED`, and the Orrery module's graph jobs stay queued with `credential_unsendable`. Permanent HTTP client errors when registering a run or uploading events also keep only the affected producer's records local; HTTP 408, 429, server errors and network failures remain retryable. HTTP 401 invalidates the short-lived credential for renewal. The production destination belongs to the package: by default the collector's production hostname (`PRODUCTION_COLLECTOR_URL`). A module's `collector_origin` setting can choose only among the package's trusted collector origins, `TRUSTED_COLLECTOR_ORIGINS`: the hostname and the Cloud Run address of the same production service (`CLOUD_RUN_COLLECTOR_URL`, the earlier default). No interface release passes `collector_origin` today, so rolling back the hostname needs a new emitter release. Development overrides accept loopback addresses only, and no environment variable changes the destination. +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. Missing or rejected credentials permanently mark that producer's events as local-only. A login saved by `hf auth login` that this user may not read, or that is not text, when neither `HF_TOKEN` nor `HUGGINGFACE_TOKEN` is set, does not make runs local-only: the login exists and can be repaired. Every pending run stays pending with the reason `huggingface_login_unreadable` (in the run's `local_only_reason` column), and one warning names only the error's type, never the file or its contents. The login is read again with the same backoff as after a collector outage, and the shutdown drain ends at once. As soon as this emitter can read the login again, whatever the collector then answers, it clears the reason of every run that waited for it, including runs it had already registered, and its shutdown drain retries a collector outage as usual; it then delivers the runs. A later emitter that can read the login delivers them too, and clears the reason when it registers each run. Other errors from huggingface_hub, such as a failed OIDC token exchange while `HF_OIDC_RESOURCE` is set, do not count as an unreadable login. The emitter sends a credential only when it is visible ASCII (letters, digits and punctuation), as every Hugging Face credential and collector token is. It refuses any other value, such as one with a pasted space, a tab, a line break or a non-ASCII character, before sending it: a bearer token (RFC 6750) cannot contain those characters, so the collector would refuse it, and Python's HTTP client raises an error that quotes the whole header value for a line break. Such a credential counts as rejected: the telemetry module marks the producer's events local-only with a warning that the credential contains characters that cannot be sent, collector submissions end `failed` with `CREDENTIAL_REJECTED`, and the Orrery module's graph jobs stay queued with `credential_unsendable`. Permanent HTTP client errors when registering a run or uploading events also keep only the affected producer's records local; HTTP 408, 429, server errors and network failures remain retryable, with backoff; a network failure includes an answer that is not valid HTTP or ends early, such as a truncated chunked answer. HTTP 401 invalidates the short-lived credential for renewal. The production destination belongs to the package: by default the collector's production hostname (`PRODUCTION_COLLECTOR_URL`). A module's `collector_origin` setting can choose only among the package's trusted collector origins, `TRUSTED_COLLECTOR_ORIGINS`: the hostname and the Cloud Run address of the same production service (`CLOUD_RUN_COLLECTOR_URL`, the earlier default). No interface release passes `collector_origin` today, so rolling back the hostname needs a new emitter release. Development overrides accept loopback addresses only, and no environment variable changes the destination. The queue upgrades the existing `20261007_01` Alembic revision to `20261008_02`, adding graph-job records without changing producer sequences, event IDs or upload eligibility. Unversioned databases are rejected without schema modification. There is no raw SQL, automatic schema adoption, or parallel legacy implementation. The hosted telemetry collector and calibration dashboard do not change. @@ -56,9 +56,9 @@ A build on Modal runs without the operator's Hugging Face credential, so its emi | 2 | A usage error, such as a lifetime below 1 or a `--development-collector-url` that is not a loopback address; nothing was sent | | 3 | No Hugging Face credential (`HF_TOKEN`, `HUGGINGFACE_TOKEN` or `hf auth login`), or a saved login this user cannot read; nothing was sent | | 4 | The collector refused the credential (401, or 403 for a user outside the PolicyEngine organization), or the credential contains characters other than visible ASCII, such as a space or a line break, which cannot be sent as a bearer token, and was not sent | - | 5 | The collector was unreachable, its answer was not valid HTTP, or its answer had no non-empty `access_token` and positive integer `expires_in` | + | 5 | The collector was unreachable, its answer was not valid HTTP, or its answer had no `access_token` of visible ASCII characters, without spaces or line breaks, and positive integer `expires_in` | | 6 | The collector answered 422: it does not accept `lifetime_seconds`, so it predates tokens for other machines | - | 7 | An unexpected error; only the error's type is printed | + | 7 | An unexpected error; only the error's type is printed. This includes a failed OIDC token exchange while `HF_OIDC_RESOURCE` is set, such as an `HfHubHTTPError` from the Hub or an `OIDCError`: it fails before the collector is contacted, so it is not 5, and it is not a missing or unreadable login, so not 3 | - **On the other machine:** the launcher sets `MICROCOSM_COLLECTOR_TOKEN`, `MICROCOSM_COLLECTOR_TOKEN_EXPIRES_AT` and `MICROCOSM_COLLECTOR_ORIGIN` from that output. With them, `CollectorSession` sends that token and never reads or exchanges a Hugging Face credential; a session for another scope, such as collector submissions, has no credential. None of the variables chooses where the emitter sends: `MICROCOSM_COLLECTOR_ORIGIN` is only compared with the origin the module settings resolve. - **Where the token goes:** only to that collector origin. The telemetry module sends it with run registrations and events. The Orrery module gets its credential through the same session (`shared_session`), but sends it to its `publication_origin`; unless that is the collector origin, which it is not by default, the session withholds a given token from Orrery on purpose, and Orrery's jobs stay queued with `credential_withheld_from_destination`. diff --git a/changelog.d/telemetry-keeps-running.fixed.md b/changelog.d/telemetry-keeps-running.fixed.md new file mode 100644 index 00000000..07c81d15 --- /dev/null +++ b/changelog.d/telemetry-keeps-running.fixed.md @@ -0,0 +1 @@ +The telemetry module no longer stops in two cases. A login saved by `hf auth login` that this user may not read, or that is not text, with neither `HF_TOKEN` nor `HUGGINGFACE_TOKEN` set, keeps runs pending with the reason `huggingface_login_unreadable`, with one warning that names only the error's type; the login is read again with backoff, and once it can be read the runs are delivered. As soon as the login can be read again, the reason is cleared, including for runs already registered, and the shutdown drain retries a collector outage as usual. Other errors from huggingface_hub, such as a failed OIDC token exchange while `HF_OIDC_RESOURCE` is set, are not reported as an unreadable login: the telemetry module handles them as before, and `microcosm-emitter telemetry-token` now exits 7 (an unexpected error, naming its type) instead of 3, because the exchange fails before the collector is contacted. A collector answer that is not valid HTTP or ends early, such as a truncated chunked answer, during run registration or event upload is retried with backoff like other network failures, and the run's state does not change. The token exchange also refuses a collector `access_token` with characters other than visible ASCII, such as a space, as an answer without a usable token, so `microcosm-emitter telemetry-token` exits 5 instead of printing a token that the emitter given it would refuse, and a workstation's telemetry module retries the exchange instead of stopping. diff --git a/src/microcosm_emitter/cli.py b/src/microcosm_emitter/cli.py index 3cdb1790..a1657673 100644 --- a/src/microcosm_emitter/cli.py +++ b/src/microcosm_emitter/cli.py @@ -33,7 +33,9 @@ #: Exit codes of `telemetry-token`, one per kind of failure; 2 is a usage #: error, as for every command, and 7 an unexpected error, which the others do -#: not name. +#: not name. A failed OIDC token exchange while HF_OIDC_RESOURCE is set is 7: +#: it fails at the CI provider or the Hub before the collector is contacted, +#: so it is not 5, and it is not a missing or unreadable login, so not 3. EXIT_MISSING_CREDENTIAL: Final = 3 EXIT_REJECTED_CREDENTIAL: Final = 4 EXIT_COLLECTOR_UNAVAILABLE: Final = 5 @@ -71,7 +73,8 @@ ), } #: `huggingface_hub` could not read the login `hf auth login` saved, such as a -#: file this user may not read, and neither variable is set. +#: file this user may not read, and neither variable is set +#: (``UnreadableLoginError``). UNREADABLE_LOGIN_MESSAGE: Final = ( "The Hugging Face login saved by `hf auth login` could not be read " "({error_type}), and neither HF_TOKEN nor HUGGINGFACE_TOKEN is set; nothing " @@ -139,6 +142,7 @@ def telemetry_token(arguments: argparse.Namespace) -> int: """ from microcosm_emitter.core.auth import ( ExchangeFailure, + UnreadableLoginError, exchange_huggingface_credential, huggingface_token, ) @@ -147,9 +151,9 @@ def telemetry_token(arguments: argparse.Namespace) -> int: origin = arguments.collector_origin or _collector_origin(None) try: credential = huggingface_token() - except (OSError, UnicodeDecodeError) as error: + except UnreadableLoginError as error: print( - UNREADABLE_LOGIN_MESSAGE.format(error_type=type(error).__name__), + UNREADABLE_LOGIN_MESSAGE.format(error_type=error.error_type), file=sys.stderr, ) return EXIT_MISSING_CREDENTIAL diff --git a/src/microcosm_emitter/core/auth.py b/src/microcosm_emitter/core/auth.py index e5c4dc8c..53b09708 100644 --- a/src/microcosm_emitter/core/auth.py +++ b/src/microcosm_emitter/core/auth.py @@ -56,6 +56,12 @@ #: http.client raises for a line break with an error that quotes the whole #: header value, and for a character outside Latin-1 with one that holds it. _SENDABLE_CREDENTIAL = re.compile(r"[!-~]+") +#: What ``_http_post`` raises when it gets no usable answer, which every caller +#: handles as a network failure: no answer or a timeout (``OSError``, +#: ``TimeoutError``), or an answer that is not valid HTTP or ends early, such +#: as a truncated chunked answer (http.client's ``HTTPException``, which is not +#: an ``OSError``). +NETWORK_FAILURES = (OSError, TimeoutError, HTTPException) class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): @@ -132,14 +138,65 @@ def _http_post( return error.code, _decode_response(body) +class UnreadableLoginError(OSError): + """The login saved by `hf auth login` could not be read, such as a file + this user may not read or one that is not text, and neither `HF_TOKEN` + nor `HUGGINGFACE_TOKEN` is set. + + ``error_type`` is the name of the error ``huggingface_hub`` raised. That + error is not kept, not even as the context of this one: its text names + the file, and a ``UnicodeDecodeError`` holds the file's bytes. An + ``OSError``, as reading a file that fails is, so callers that handled the + ``PermissionError`` before still handle it. + """ + + def __init__(self, error_type: str) -> None: + super().__init__( + f"the saved Hugging Face login could not be read ({error_type})" + ) + self.error_type = error_type + + +def _saved_login_unreadable(error: OSError | UnicodeDecodeError) -> bool: + """Whether ``get_token()`` raised ``error`` reading the login files that + `hf auth login` saved: a ``PermissionError``, bytes that are not text, or + another ``OSError`` that names a file, such as ``IsADirectoryError``. + + With ``HF_OIDC_RESOURCE`` set, huggingface_hub exchanges a CI OIDC token + instead of reading those files, so none of its errors count, such as the + ``HfHubHTTPError`` (an ``OSError`` that names no file) of a 503 from the + Hub, or an answer that is not text. + """ + if os.environ.get("HF_OIDC_RESOURCE"): + return False + return isinstance(error, (PermissionError, UnicodeDecodeError)) or ( + isinstance(error, OSError) and error.filename is not None + ) + + def huggingface_token() -> str | None: """The build's Hugging Face token: `HF_TOKEN`, then `HUGGINGFACE_TOKEN`, then - the token saved by `hf auth login`; `None` without one.""" - return ( + the token saved by `hf auth login`; `None` without one. + + Raises ``UnreadableLoginError`` when only the saved login is left and it + cannot be read (``_saved_login_unreadable``); every other error of + ``get_token()`` propagates unchanged. + """ + variable_token = ( os.environ.get("HF_TOKEN", "").strip() or os.environ.get("HUGGINGFACE_TOKEN", "").strip() - or get_token() ) + if variable_token: + return variable_token + try: + return get_token() + except (OSError, UnicodeDecodeError) as error: + # huggingface_hub catches only FileNotFoundError when it reads the file. + if not _saved_login_unreadable(error): + raise + error_type = type(error).__name__ + # Outside the except block, so the new error has no context to keep. + raise UnreadableLoginError(error_type) @dataclass(frozen=True) @@ -201,9 +258,10 @@ def given_collector_token( @dataclass(frozen=True) class ExchangedToken: - """A usable answer to an exchange: a 200 with a non-empty ``access_token`` - and a positive whole number of seconds, ``expires_in``. Its repr leaves - the token out.""" + """A usable answer to an exchange: a 200 with an ``access_token`` of + visible ASCII characters, which can be sent as a bearer token, and a + positive whole number of seconds, ``expires_in``. Its repr leaves the + token out.""" access_token: str = field(repr=False) expires_in: int @@ -246,8 +304,11 @@ def exchange_huggingface_credential( Without ``scope`` the body is ``{}``, which the collector answers with a ``telemetry:write`` token. ``lifetime_seconds`` asks for a token that lasts that long, for a build on a machine without the credential; the - collector may grant less. A usable answer is a 200 with a non-empty - ``access_token`` and a positive integer ``expires_in``. + collector may grant less. A usable answer is a 200 with an + ``access_token`` that ``_SENDABLE_CREDENTIAL`` accepts, as the Hugging + Face credential must be, and a positive integer ``expires_in``. Any other + token could not be sent as a bearer token, so it counts as no usable + token: ``unavailable``. """ if not hugging_face_token: @@ -261,8 +322,7 @@ def exchange_huggingface_credential( status, answer = (post or _http_post)( origin + TOKEN_EXCHANGE_PATH, body, hugging_face_token ) - except (OSError, TimeoutError, HTTPException): - # HTTPException: an answer that is not valid HTTP, or is truncated. + except NETWORK_FAILURES: return ExchangeFailure("unavailable") if status in {HTTPStatus.UNAUTHORIZED, HTTPStatus.FORBIDDEN}: return ExchangeFailure("rejected", status) @@ -280,7 +340,7 @@ def exchange_huggingface_credential( if ( status != HTTPStatus.OK or not isinstance(access_token, str) - or not access_token + or not _SENDABLE_CREDENTIAL.fullmatch(access_token) or type(expires_in) is not int or expires_in <= 0 ): @@ -354,6 +414,12 @@ class CollectorSession: ``token_expired`` or ``token_rejected``. A given token obtained from another origin than ``origin`` raises ``GivenCollectorTokenError``: its origin is only compared, and never chooses where anything is sent. + + Without a given token, ``credential()`` raises the ``UnreadableLoginError`` + of ``huggingface_token`` when the saved login cannot be read; the + telemetry module keeps its runs pending with the reason + ``huggingface_login_unreadable`` and reads the login again after a + backoff. """ def __init__( diff --git a/src/microcosm_emitter/core/constants.py b/src/microcosm_emitter/core/constants.py index 6754ffe5..1035dfb7 100644 --- a/src/microcosm_emitter/core/constants.py +++ b/src/microcosm_emitter/core/constants.py @@ -68,11 +68,23 @@ # already has; a successful registration clears it. PENDING_GIVEN_TOKEN_EXPIRED: Final = "collector_token_expired" PENDING_GIVEN_TOKEN_REJECTED: Final = "collector_token_rejected" +# Why a pending run waits: the login saved by `hf auth login` could not be read, +# such as a file this user may not read, and neither HF_TOKEN nor +# HUGGINGFACE_TOKEN is set. The login exists and can be repaired, so the run +# stays pending for this emitter, or a later one, to deliver once it can be read. +PENDING_HUGGINGFACE_LOGIN_UNREADABLE: Final = "huggingface_login_unreadable" NO_CREDENTIAL_MESSAGE: Final = ( "Microcosm telemetry is local-only: no ambient Hugging Face credential was " "found. The dataset build will continue." ) +# Only the error's type, since its text names the file, or quotes its bytes. +UNREADABLE_LOGIN_WAITING_MESSAGE: Final = ( + "Microcosm telemetry is waiting: the Hugging Face login saved by `hf auth " + "login` could not be read ({error_type}), and neither HF_TOKEN nor " + "HUGGINGFACE_TOKEN is set. Queued events stay pending, and are delivered once " + "the login can be read. The dataset build will continue." +) REJECTED_CREDENTIAL_MESSAGE: Final = ( "Microcosm telemetry is local-only for this run: the ambient Hugging Face " "credential was not accepted as a PolicyEngine organization member. The " diff --git a/src/microcosm_emitter/core/models.py b/src/microcosm_emitter/core/models.py index 5ef2969f..c4d8f44e 100644 --- a/src/microcosm_emitter/core/models.py +++ b/src/microcosm_emitter/core/models.py @@ -87,7 +87,8 @@ class TelemetryRunRecord(SpoolModel): default=UPLOAD_STATE_PENDING, ) #: Why a local-only run is excluded from upload or, for a pending run, - #: why delivery waits (a given collector token expired or was refused). + #: why delivery waits: a given collector token expired or was refused, or + #: the saved Hugging Face login could not be read. local_only_reason: Mapped[str | None] = mapped_column(Text) updated_at: Mapped[str] = mapped_column(Text, nullable=False) events: Mapped[list[TelemetryEventRecord]] = relationship( diff --git a/src/microcosm_emitter/telemetry/service/collector.py b/src/microcosm_emitter/telemetry/service/collector.py index 79279d19..e59d7564 100644 --- a/src/microcosm_emitter/telemetry/service/collector.py +++ b/src/microcosm_emitter/telemetry/service/collector.py @@ -9,8 +9,10 @@ from typing import Any from microcosm_emitter.core.auth import ( + NETWORK_FAILURES, CollectorSession, SessionResult, + UnreadableLoginError, _http_post, huggingface_token, ) @@ -26,9 +28,11 @@ NO_CREDENTIAL_MESSAGE, PENDING_GIVEN_TOKEN_EXPIRED, PENDING_GIVEN_TOKEN_REJECTED, + PENDING_HUGGINGFACE_LOGIN_UNREADABLE, REJECTED_CREDENTIAL_MESSAGE, RUN_EVENTS_PATH_TEMPLATE, RUN_REGISTRATION_PATH, + UNREADABLE_LOGIN_WAITING_MESSAGE, UNSENDABLE_CREDENTIAL_MESSAGE, ) from microcosm_emitter.telemetry.constants import ( @@ -79,14 +83,25 @@ def __init__( ) self._registered: set[str] = set() self._warned_no_token = False - #: Set once the given collector token can no longer be used: nothing - #: more can be delivered until an emitter with a credential runs. - self.waiting_for_credential = False + self._warned_unreadable_login = False + #: Set once the given collector token can no longer be used. + self._given_token_unusable = False + #: The runs, as (run ID, producer ID), recorded as waiting for the + #: saved Hugging Face login to be readable; empty while it can be read. + self._waiting_for_login: set[tuple[str, str]] = set() self._recorded_waiting: set[str] = set() self._warned_denied: set[str] = set() self._next_attempt_at = 0.0 self._retry_seconds = INITIAL_RETRY_SECONDS + @property + def waiting_for_credential(self) -> bool: + """Whether nothing can be delivered from this emitter, so its shutdown + drain ends at once: the given collector token can no longer be used, + or the saved Hugging Face login could not be read at the last attempt. + """ + return self._given_token_unusable or bool(self._waiting_for_login) + def flush_once(self) -> bool: """Attempt one delivery pass without waiting for retry deadlines.""" @@ -111,7 +126,7 @@ def flush_once(self) -> bool: {"events": events}, token, ) - except (OSError, TimeoutError): + except NETWORK_FAILURES: self._defer_retry() continue if status == HTTPStatus.ACCEPTED: @@ -136,7 +151,15 @@ def flush_once(self) -> bool: return made_progress def _collector_token(self, registration: Mapping[str, Any]) -> str | None: - result, token = self.session.credential() + try: + result, token = self.session.credential() + except UnreadableLoginError as error: + self._wait_for_readable_login(registration, error.error_type) + return None + if self._waiting_for_login: + # The session raised until it could read the login, so it can now, + # whatever the exchange answered. + self._end_wait_for_readable_login() if result == "missing": if not self._warned_no_token: print(NO_CREDENTIAL_MESSAGE, file=sys.stderr, flush=True) @@ -156,12 +179,45 @@ def _collector_token(self, registration: Mapping[str, Any]) -> str | None: self._defer_retry() return token + def _wait_for_readable_login( + self, registration: Mapping[str, Any], error_type: str + ) -> None: + """Keep the run pending while the saved login cannot be read. + + Pending, not local-only, which cannot be undone and would also apply + to every other pending run in the event queue file: the login exists + and can be repaired, and then this emitter or a later one delivers + the run. The login is read again after the backoff, as the exchange + is after an outage. The warning names only the error's type, once. + """ + + if not self._warned_unreadable_login: + print( + UNREADABLE_LOGIN_WAITING_MESSAGE.format(error_type=error_type), + file=sys.stderr, + flush=True, + ) + self._warned_unreadable_login = True + run = (str(registration["run_id"]), str(registration["producer_id"])) + if run not in self._waiting_for_login: + self.spool.record_pending_reason(*run, PENDING_HUGGINGFACE_LOGIN_UNREADABLE) + self._waiting_for_login.add(run) + self._defer_retry() + + def _end_wait_for_readable_login(self) -> None: + """Clear the reason of every run that waited for the saved login, + registered or not, so that the shutdown drain retries as usual.""" + + for run_id, producer_id in self._waiting_for_login: + self.spool.record_pending_reason(run_id, producer_id, None) + self._waiting_for_login.clear() + def _keep_pending(self, registration: Mapping[str, Any], reason: str) -> None: """Leave the run for an emitter with a credential, saying why once.""" - if not self.waiting_for_credential: + if not self._given_token_unusable: print(GIVEN_TOKEN_UNUSABLE_MESSAGE, file=sys.stderr, flush=True) - self.waiting_for_credential = True + self._given_token_unusable = True registration_key = _registration_key(registration) if registration_key not in self._recorded_waiting: self.spool.record_pending_reason( @@ -177,13 +233,14 @@ def _register(self, registration: Mapping[str, Any], token: str) -> bool: registration, token, ) - except (OSError, TimeoutError): + except NETWORK_FAILURES: self._defer_retry() return False if status == HTTPStatus.CREATED: self._registered.add(registration_key) self._retry_seconds = INITIAL_RETRY_SECONDS - # A run left by an emitter whose given token became unusable. + # A run left by an emitter whose given token became unusable, or + # that could not read its saved login. self.spool.record_pending_reason( str(registration["run_id"]), str(registration["producer_id"]), None ) diff --git a/tests/telemetry/test_delivery.py b/tests/telemetry/test_delivery.py index 4bd55abc..47917108 100644 --- a/tests/telemetry/test_delivery.py +++ b/tests/telemetry/test_delivery.py @@ -1,9 +1,13 @@ +import errno import json +import os import threading +import time from datetime import timedelta from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from types import SimpleNamespace +import httpx import pytest from alembic import command from given_token import ( @@ -12,6 +16,7 @@ given_token_variables, set_variables, ) +from huggingface_hub.errors import HfHubHTTPError from sqlalchemy.orm import Session from microcosm_emitter.core import auth as auth_module @@ -31,6 +36,7 @@ TelemetryRunRecord, serialized_json_length, ) +from microcosm_emitter.host.contracts import ModuleContext from microcosm_emitter.telemetry.client import ( TelemetryRun, ) @@ -41,6 +47,7 @@ from microcosm_emitter.telemetry.service import resources as resources_module from microcosm_emitter.telemetry.service import spool as spool_module from microcosm_emitter.telemetry.service.collector import CollectorDelivery +from microcosm_emitter.telemetry.service.module import TelemetryModule from microcosm_emitter.telemetry.service.sanitization import ( sanitize_details, sanitize_json, @@ -71,6 +78,40 @@ def refuse() -> str: monkeypatch.setattr(auth_module, "get_token", refuse) +#: Collector answers that are not valid HTTP. http.client raises an +#: HTTPException for each, which is not an OSError. +MALFORMED_ANSWERS = { + # Ends inside its first chunk: IncompleteRead. + "truncated chunked answer": ( + b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n" + b'Transfer-Encoding: chunked\r\n\r\n40\r\n{"accepted"' + ), + # BadStatusLine. + "answer that is not HTTP": b"not an HTTP answer\r\n", +} + + +def _telemetry_module(tmp_path, collector_url: str) -> TelemetryModule: + """A telemetry module for one run, started as the host starts it.""" + module = TelemetryModule( + { + "registration": _registration(), + "spool_path": str(tmp_path / "events.sqlite3"), + "development_collector_url": collector_url, + }, + ModuleContext(os.getpid()), + ) + module.initialize() + return module + + +def _run_state(spool: EventSpool, run_id: str = "run-a") -> tuple[str, str | None]: + """The run's upload state and the reason it records.""" + with Session(spool._engine) as session: + record = session.get(TelemetryRunRecord, (run_id, "producer-a")) + return record.upload_state, record.local_only_reason + + def _event(stage_id: str = "compile_targets") -> dict[str, object]: return { "timestamp": "2026-10-02T10:00:00+00:00", @@ -499,6 +540,311 @@ def test_missing_credential_never_contacts_collector_and_stays_local_only( assert reopened.pending_runs() == [] +@pytest.mark.parametrize( + "unreadable_login", + [ + PermissionError(13, "Permission denied", "/synthetic/hf/token"), + UnicodeDecodeError("utf-8", b"\xffsynthetic", 0, 1, "invalid start byte"), + ], + ids=["not readable by this user", "not text"], +) +def test_an_unreadable_saved_login_keeps_runs_pending_until_it_can_be_read( + tmp_path, monkeypatch, capsys, unreadable_login +) -> None: + # An earlier build's run, still pending in the event queue file. + spool_path = tmp_path / "events.sqlite3" + earlier = _registration("run-earlier") + earlier_queue = EventSpool(spool_path) + earlier_queue.register(earlier) + earlier_queue.append(earlier, _event()) + earlier_queue.close() + # Neither HF_TOKEN nor HUGGINGFACE_TOKEN is set, so the emitter reads the + # login saved by `hf auth login`. + login_reads: list[str] = [] + + def read_saved_login() -> str: + login_reads.append("read") + raise unreadable_login + + def refuse_request(*arguments, **options): + raise AssertionError("sent a request without a credential") + + monkeypatch.setattr(auth_module, "get_token", read_saved_login) + monkeypatch.setattr(auth_module, "_http_post", refuse_request) + monkeypatch.setattr(collector_module, "_http_post", refuse_request) + real_monotonic = time.monotonic + clock = [100.0] + monkeypatch.setattr(collector_module.time, "monotonic", lambda: clock[0]) + module = _telemetry_module(tmp_path, DEVELOPMENT_COLLECTOR_URL) + try: + # The host stops a module whose tick raises. + module.tick(clock[0]) + + # Both runs stay pending, saying why; neither becomes local-only. + assert len(module.spool.pending_runs()) == 2 + for run_id in ("run-a", "run-earlier"): + assert _run_state(module.spool, run_id) == ( + "pending", + "huggingface_login_unreadable", + ) + + # The login is read again only after the backoff. + reads = len(login_reads) + module.tick(clock[0]) + assert len(login_reads) == reads + clock[0] += 100.0 + module.tick(clock[0]) + assert len(login_reads) > reads + finally: + monkeypatch.setattr(collector_module.time, "monotonic", real_monotonic) + started = time.monotonic() + module.close(started + 5) + # Nothing can be delivered from here, so the shutdown drain ends at once. + assert time.monotonic() - started < 1 + warning = capsys.readouterr().err + assert warning.count("Microcosm telemetry is waiting") == 1 + # Only the error's type: its text names the path or quotes the file. + assert f"could not be read ({type(unreadable_login).__name__})" in warning + assert "synthetic" not in warning and "0xff" not in warning + + # Once the login can be read, the next emitter delivers both runs. + def collector(url, payload, token, *, timeout=5.0): + if url.endswith("/exchange"): + return 200, {"access_token": "synthetic-session", "expires_in": 3600} + assert token == "synthetic-session" + return (201, {}) if url.endswith("/v1/runs") else (202, {}) + + monkeypatch.setattr(auth_module, "get_token", lambda: "hf_synthetic_member") + monkeypatch.setattr(collector_module, "_http_post", collector) + reopened = EventSpool(spool_path) + try: + assert CollectorDelivery( + reopened, development_collector_url=DEVELOPMENT_COLLECTOR_URL + ).flush_once() + assert not reopened.has_pending() + for run_id in ("run-a", "run-earlier"): + assert _run_state(reopened, run_id) == ("pending", None) + finally: + reopened.close() + + +def test_delivery_resumes_once_the_saved_login_can_be_read( + tmp_path, monkeypatch +) -> None: + spool = EventSpool(tmp_path / "events.sqlite3") + registration = _registration() + spool.register(registration) + spool.append(registration, _event()) + saved_login: list[str | Exception] = [ + PermissionError(13, "Permission denied", "/synthetic/hf/token") + ] + + def read_saved_login() -> str: + if isinstance(saved_login[0], Exception): + raise saved_login[0] + return saved_login[0] + + def collector(url, payload, token, *, timeout=5.0): + if url.endswith("/exchange"): + return 200, {"access_token": "synthetic-session", "expires_in": 3600} + return (201, {}) if url.endswith("/v1/runs") else (202, {}) + + monkeypatch.setattr(auth_module, "get_token", read_saved_login) + monkeypatch.setattr(collector_module, "_http_post", collector) + clock = [100.0] + monkeypatch.setattr(collector_module.time, "monotonic", lambda: clock[0]) + delivery = CollectorDelivery( + spool, development_collector_url=DEVELOPMENT_COLLECTOR_URL + ) + try: + assert not delivery.flush_once() + assert delivery.waiting_for_credential + + # The user repairs the file, and the next attempt delivers the run. + saved_login[0] = "hf_synthetic_member" + clock[0] += 100.0 + assert delivery.flush_once() + assert not spool.has_pending() + assert _run_state(spool) == ("pending", None) + # So the shutdown drain waits for deliveries again. + assert not delivery.waiting_for_credential + finally: + spool.close() + + +def test_a_registered_run_loses_the_unreadable_login_reason_once_it_can_be_read( + tmp_path, monkeypatch +) -> None: + spool = EventSpool(tmp_path / "events.sqlite3") + registration = _registration() + spool.register(registration) + spool.append(registration, _event()) + saved_login: list[str | Exception] = ["hf_synthetic_member"] + + def read_saved_login() -> str: + if isinstance(saved_login[0], Exception): + raise saved_login[0] + return saved_login[0] + + def collector(url, payload, token, *, timeout=5.0): + if url.endswith("/exchange"): + return 200, {"access_token": "synthetic-session", "expires_in": 3600} + return (201, {}) if url.endswith("/v1/runs") else (202, {}) + + monkeypatch.setattr(auth_module, "get_token", read_saved_login) + monkeypatch.setattr(collector_module, "_http_post", collector) + clock = [100.0] + monkeypatch.setattr(collector_module.time, "monotonic", lambda: clock[0]) + delivery = CollectorDelivery( + spool, development_collector_url=DEVELOPMENT_COLLECTOR_URL + ) + try: + # This emitter registers the run and delivers its first event. + assert delivery.flush_once() + spool.append(registration, _event("load_inputs")) + + # When the exchanged token is renewed, the login cannot be read. + saved_login[0] = PermissionError(13, "Permission denied", "/synthetic/hf/token") + clock[0] += 3600.0 + assert not delivery.flush_once() + assert _run_state(spool) == ("pending", "huggingface_login_unreadable") + + # Once it can be read, the reason goes with the delivery. + saved_login[0] = "hf_synthetic_member" + clock[0] += 100.0 + assert delivery.flush_once() + assert not spool.has_pending() + assert _run_state(spool) == ("pending", None) + finally: + spool.close() + + +def test_a_collector_outage_after_the_login_can_be_read_is_retried_at_shutdown( + tmp_path, monkeypatch +) -> None: + spool = EventSpool(tmp_path / "events.sqlite3") + registration = _registration() + spool.register(registration) + spool.append(registration, _event()) + saved_login: list[str | Exception] = [ + PermissionError(13, "Permission denied", "/synthetic/hf/token") + ] + + def read_saved_login() -> str: + if isinstance(saved_login[0], Exception): + raise saved_login[0] + return saved_login[0] + + monkeypatch.setattr(auth_module, "get_token", read_saved_login) + monkeypatch.setattr( + collector_module, + "_http_post", + lambda *arguments, **options: (503, {"detail": "temporarily unavailable"}), + ) + clock = [100.0] + monkeypatch.setattr(collector_module.time, "monotonic", lambda: clock[0]) + delivery = CollectorDelivery( + spool, development_collector_url=DEVELOPMENT_COLLECTOR_URL + ) + try: + assert not delivery.flush_once() + assert delivery.waiting_for_credential + + # The login can be read again, but the collector does not answer. + saved_login[0] = "hf_synthetic_member" + clock[0] += 100.0 + assert not delivery.flush_once() + + # So the shutdown drain retries, as for any outage, and the run no + # longer says it waits for the login. + assert not delivery.waiting_for_credential + assert _run_state(spool) == ("pending", None) + assert spool.pending_runs() == [registration] + finally: + spool.close() + + +@pytest.mark.parametrize( + "error", + [ + PermissionError(errno.EACCES, "Permission denied", "/synthetic/hf/token"), + IsADirectoryError(errno.EISDIR, "Is a directory", "/synthetic/hf/token"), + OSError( + errno.ELOOP, "Too many levels of symbolic links", "/synthetic/hf/token" + ), + UnicodeDecodeError("utf-8", b"\xffsynthetic", 0, 1, "invalid start byte"), + ], + ids=[ + "not readable by this user", + "a directory", + "a symbolic link loop", + "not text", + ], +) +def test_a_saved_login_that_cannot_be_read_raises_an_error_without_its_details( + monkeypatch, error +) -> None: + def get_token() -> str: + raise error + + monkeypatch.setattr(auth_module, "get_token", get_token) + + with pytest.raises(auth_module.UnreadableLoginError) as raised: + auth_module.huggingface_token() + + unreadable = raised.value + assert unreadable.error_type == type(error).__name__ + # Nothing keeps the original error, not even as context: its text names + # the file, and a UnicodeDecodeError holds the file's bytes. + assert unreadable.__context__ is None and unreadable.__cause__ is None + assert "synthetic" not in str(unreadable) + + +def _hub_outage() -> HfHubHTTPError: + """What huggingface_hub raises for a 503 from the Hub: an OSError.""" + request = httpx.Request("POST", "https://huggingface.co/oauth/token") + return HfHubHTTPError( + "503 Server Error: Service Unavailable", + response=httpx.Response(503, request=request), + ) + + +@pytest.mark.parametrize( + ("oidc_resource", "error"), + [ + ("datasets/policyengine/synthetic", _hub_outage()), + ( + "datasets/policyengine/synthetic", + UnicodeDecodeError("utf-8", b"\xff", 0, 1, "invalid start byte"), + ), + (None, _hub_outage()), + ], + ids=[ + "Hub outage during the OIDC exchange", + "OIDC answer that is not text", + "Hub outage without OIDC", + ], +) +def test_only_a_saved_login_that_cannot_be_read_counts_as_unreadable( + monkeypatch, oidc_resource, error +) -> None: + # With HF_OIDC_RESOURCE set, huggingface_hub exchanges a CI OIDC token + # instead of reading the saved login. Its errors, and any OSError that + # does not come from reading a file, propagate as before. + if oidc_resource is not None: + monkeypatch.setenv("HF_OIDC_RESOURCE", oidc_resource) + + def get_token() -> str: + raise error + + monkeypatch.setattr(auth_module, "get_token", get_token) + + with pytest.raises(type(error)) as raised: + auth_module.huggingface_token() + + assert raised.value is error + + @pytest.mark.parametrize("endpoint", ["registration", "events"]) @pytest.mark.parametrize("status", [400, 404, 413, 422]) def test_permanent_collector_rejection_retains_only_rejected_producer( @@ -609,6 +955,120 @@ def post(url, payload, token): spool.close() +@pytest.mark.parametrize("malformed_answer", sorted(MALFORMED_ANSWERS)) +@pytest.mark.parametrize("endpoint", ["registration", "events"]) +def test_a_malformed_collector_answer_is_retried_like_a_network_failure( + tmp_path, monkeypatch, endpoint, malformed_answer +) -> None: + malformed_path = ( + "/v1/runs" if endpoint == "registration" else "/v1/runs/run-a/events" + ) + paths: list[str] = [] + + class Collector(BaseHTTPRequestHandler): + """Answers the first request to ``malformed_path`` with broken HTTP.""" + + def do_POST(self): + self.rfile.read(int(self.headers["Content-Length"])) + paths.append(self.path) + if self.path.endswith("/exchange"): + return self._json( + 200, {"access_token": "synthetic-session", "expires_in": 3600} + ) + if self.path == malformed_path and paths.count(self.path) == 1: + self.wfile.write(MALFORMED_ANSWERS[malformed_answer]) + return + self._json(201 if self.path == "/v1/runs" else 202, {}) + + def _json(self, status: int, document: dict[str, object]) -> None: + body = json.dumps(document).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, format, *args): + return + + server = ThreadingHTTPServer(("127.0.0.1", 0), Collector) + server_thread = threading.Thread(target=server.serve_forever) + server_thread.start() + clock = [100.0] + monkeypatch.setattr(collector_module.time, "monotonic", lambda: clock[0]) + monkeypatch.setenv("HF_TOKEN", "hf_synthetic_member") + module = _telemetry_module(tmp_path, f"http://127.0.0.1:{server.server_port}") + try: + # The host stops a module whose tick raises. + module.tick(clock[0]) + assert paths.count(malformed_path) == 1 + + # The run and its queued event stay as they were. + assert module.spool.pending_runs() == [module.registration] + assert _run_state(module.spool) == ("pending", None) + assert len(module.spool.batch("run-a", "producer-a")) == 1 + + # Retried after the backoff, as for any network failure. + module.tick(clock[0]) + assert paths.count(malformed_path) == 1 + clock[0] += 100.0 + module.tick(clock[0]) + assert paths.count(malformed_path) == 2 + assert not module.spool.has_pending() + finally: + module.close(clock[0]) + server.shutdown() + server_thread.join(timeout=2) + server.server_close() + + +def test_a_collector_token_that_cannot_be_sent_is_retried_like_an_outage( + tmp_path, monkeypatch +) -> None: + exchanges: list[str] = [] + sent: list[str] = [] + + def collector_issuing_a_token_with_a_space(url, payload, token, *, timeout=5.0): + exchanges.append(url) + return 200, {"access_token": "session token", "expires_in": 3600} + + # The real request, which refuses a token it cannot send with ValueError. + real_post = collector_module._http_post + + def send(url, payload, token, *, timeout=5.0): + sent.append(url) + return real_post(url, payload, token, timeout=timeout) + + monkeypatch.setattr( + auth_module, "_http_post", collector_issuing_a_token_with_a_space + ) + monkeypatch.setattr(collector_module, "_http_post", send) + monkeypatch.setenv("HF_TOKEN", "hf_synthetic_member") + clock = [100.0] + monkeypatch.setattr(collector_module.time, "monotonic", lambda: clock[0]) + module = _telemetry_module(tmp_path, DEVELOPMENT_COLLECTOR_URL) + try: + # The host stops a module whose tick raises. + module.tick(clock[0]) + + # The session does not keep the token, so nothing is sent with it, + # and the run and its queued event stay as they were. + assert sent == [] + assert module.spool.pending_runs() == [module.registration] + assert _run_state(module.spool) == ("pending", None) + assert len(module.spool.batch("run-a", "producer-a")) == 1 + + # The exchange is tried again after the backoff, as after an outage. + module.tick(clock[0]) + assert len(exchanges) == 1 + clock[0] += 100.0 + module.tick(clock[0]) + assert len(exchanges) == 2 + assert sent == [] + finally: + module.close(clock[0]) + + def test_resource_sampler_keeps_reaped_child_cpu_monotonic(monkeypatch) -> None: state = {"child_alive": True} diff --git a/tests/test_telemetry_token_command.py b/tests/test_telemetry_token_command.py index 4c7c011f..92436046 100644 --- a/tests/test_telemetry_token_command.py +++ b/tests/test_telemetry_token_command.py @@ -8,8 +8,10 @@ import json from datetime import datetime, timedelta +import httpx import pytest from fake_collector import FakeCollector +from huggingface_hub.errors import HfHubHTTPError from microcosm_emitter import cli from microcosm_emitter.core import auth as auth_module @@ -41,6 +43,15 @@ def unreadable_saved_login() -> str: raise PermissionError("the saved token file is not readable") +def failed_oidc_exchange() -> str: + """get_token() with HF_OIDC_RESOURCE set, when the Hub answers 503.""" + request = httpx.Request("POST", "https://huggingface.co/oauth/token") + raise HfHubHTTPError( + "503 Server Error: Service Unavailable", + response=httpx.Response(503, request=request), + ) + + def unexpected_error(*arguments, **options): raise RuntimeError(f"an error quoting {SYNTHETIC_HUGGING_FACE_TOKEN}") @@ -95,6 +106,8 @@ def test_prints_a_telemetry_token_for_the_requested_lifetime( # Not a bearer token the collector would accept, so it counts as refused. ("credential with a line break", 4), ("unreadable saved login", 3), + # Not a saved login that cannot be read, so not exit 3. + ("OIDC token exchange that fails", 7), ("collector unreachable", 5), ("collector answers 503", 5), ("collector answer that is not HTTP", 5), @@ -131,6 +144,10 @@ def test_each_failure_has_its_own_exit_code_and_prints_nothing( elif failure == "unreadable saved login": monkeypatch.delenv("HF_TOKEN") monkeypatch.setattr(auth_module, "get_token", unreadable_saved_login) + elif failure == "OIDC token exchange that fails": + monkeypatch.delenv("HF_TOKEN") + monkeypatch.setenv("HF_OIDC_RESOURCE", "datasets/policyengine/synthetic") + monkeypatch.setattr(auth_module, "get_token", failed_oidc_exchange) elif failure == "unexpected error": monkeypatch.setattr( auth_module, "exchange_huggingface_credential", unexpected_error @@ -149,6 +166,8 @@ def test_each_failure_has_its_own_exit_code_and_prints_nothing( assert "cannot be sent as a bearer token" in output.err elif failure == "unreadable saved login": assert "could not be read (PermissionError)" in output.err + elif failure == "OIDC token exchange that fails": + assert "unexpected HfHubHTTPError" in output.err if expected_exit_code in {2, 3, 7} or failure == "credential with a line break": assert collector.exchange_requests == [] @@ -169,10 +188,23 @@ def test_an_exchanged_token_leaves_the_token_out_of_its_repr(collector) -> None: {"access_token": "session-token"}, {"access_token": "session-token", "expires_in": 0}, {"access_token": "session-token", "expires_in": "3600"}, + # Not sendable as a bearer token, like an unsendable Hugging Face + # credential: the emitter given it would refuse it at start. + {"access_token": "session token", "expires_in": 3600}, + {"access_token": "session-token\nsecond-part", "expires_in": 3600}, + {"access_token": "session-t\u00f6ken", "expires_in": 3600}, + ], + ids=[ + "empty token", + "no lifetime", + "zero lifetime", + "lifetime as text", + "token with a space", + "token with a line break", + "token with a non-ASCII character", ], - ids=["empty token", "no lifetime", "zero lifetime", "lifetime as text"], ) -def test_a_token_answer_without_a_token_and_a_lifetime_is_unusable( +def test_a_token_answer_without_a_usable_token_and_a_lifetime_is_unusable( collector, operator_login, capsys, answer ) -> None: collector.exchange_answer = answer