From a594d0f2d6d863e83a3fd216bf2f853a8e7831ad Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Fri, 9 Oct 2026 13:33:08 -0400 Subject: [PATCH 1/4] Keep one build's telemetry login from deciding other builds' runs Every emitter service delivered every pending run in the host's shared spool and judged each against its own Hugging Face credential, so a build started without a login (or with one the collector refused) permanently made every other build's queued run local-only. The collector gives a run to the Hugging Face user who registers it first, so a service now delivers its own run, and another producer's run only once that producer's service has exited: each service holds an advisory flock on a per-producer lease file beside the spool, which the kernel drops on any exit. A run with no lease file (an older checkout) is adopted only after 15 idle minutes. An answer about the caller's credential (none, refused, not the run's owner) decides only the caller's own run; an adopted run goes local-only only for an answer about the run itself. A refused credential is exchanged once, and the shutdown drain waits only for the build's own run. No spool migration. Co-Authored-By: Claude Opus 5.5 --- README.md | 7 +- .../telemetry-foreign-run-local-only.fixed.md | 1 + .../telemetry_emitter_service/collector.py | 272 ++++-- .../telemetry_emitter_service/constants.py | 11 + .../build/telemetry_emitter_service/leases.py | 211 +++++ .../build/telemetry_emitter_service/main.py | 17 +- .../telemetry_emitter_service/runtime.py | 9 +- .../build/telemetry_emitter_service/spool.py | 33 + .../shared/test_telemetry_emitter.py | 14 +- .../shared/test_telemetry_foreign_runs.py | 820 ++++++++++++++++++ 10 files changed, 1332 insertions(+), 63 deletions(-) create mode 100644 changelog.d/telemetry-foreign-run-local-only.fixed.md create mode 100644 packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/leases.py create mode 100644 packages/microcosm-build/tests/engine_free/shared/test_telemetry_foreign_runs.py diff --git a/README.md b/README.md index f25d358cd6..8c274ad511 100644 --- a/README.md +++ b/README.md @@ -49,8 +49,11 @@ until you restart the kernel, load it by path with Supported US and UK build commands always start the local telemetry emitter service. It reports live progress to the hosted collector when the operator's existing Hugging Face login is accepted; otherwise it retains the events -locally and the build continues. This live event path is independent of the -staging run files described below. +locally and the build continues. Every build on a machine shares one local +queue. A build's login decides only that build's own run. A run whose service +has exited is delivered later by another build's service whose login the +collector accepts for it. This live event path is independent of the staging +run files described below. US fiscal refresh builds also write pre-release staging run files **by default**. Progress JSON is uploaded to `policyengine/populace-us-staging` diff --git a/changelog.d/telemetry-foreign-run-local-only.fixed.md b/changelog.d/telemetry-foreign-run-local-only.fixed.md new file mode 100644 index 0000000000..ad2b438018 --- /dev/null +++ b/changelog.d/telemetry-foreign-run-local-only.fixed.md @@ -0,0 +1 @@ +A build's telemetry service no longer disables hosted telemetry for other builds on the same machine. Every service shares one local spool. Before, a service with no Hugging Face login, or with a login the collector refused, permanently marked every other build's queued run local-only. Now a service touches another build's run only after that build's service has exited, and only one service does so at a time. Each service holds an advisory lock file next to the spool. A run with no lock file, from an older checkout, waits until it has been idle for 15 minutes. A missing, refused or non-owning login decides only the service's own run; another build's run stays pending for a service whose login the collector accepts for it. A refused login is sent to the collector once, not on every pass, and the shutdown drain waits only for the build's own run. 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 0d7e4d294a..6e7b7b22d5 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 @@ -2,6 +2,8 @@ from __future__ import annotations +import contextlib +import hashlib import json import os import sys @@ -9,6 +11,7 @@ import urllib.error import urllib.request from collections.abc import Mapping +from datetime import UTC, datetime, timedelta from http import HTTPStatus from typing import Any from urllib.parse import urlsplit @@ -24,6 +27,7 @@ HTTP_TIMEOUT_SECONDS, HTTP_USER_AGENT, INITIAL_RETRY_SECONDS, + LEASE_SWEEP_INTERVAL_SECONDS, LOCAL_ONLY_MISSING_CREDENTIAL, LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION, LOCAL_ONLY_REJECTED_CREDENTIAL, @@ -34,6 +38,8 @@ MAX_RETRY_SECONDS, MINIMUM_TOKEN_LIFETIME_SECONDS, NO_CREDENTIAL_MESSAGE, + ORPHAN_IDLE_SECONDS, + OWN_LEASE_TIMEOUT_SECONDS, PRODUCTION_COLLECTOR_URL, REJECTED_CREDENTIAL_MESSAGE, REJECTED_EVENTS_MESSAGE, @@ -42,6 +48,7 @@ TOKEN_EXCHANGE_PATH, TOKEN_REFRESH_MARGIN_SECONDS, ) +from microcosm.build.telemetry_emitter_service.leases import ProducerLeases from microcosm.build.telemetry_emitter_service.spool import EventSpool @@ -141,14 +148,36 @@ def _registration_key(registration: Mapping[str, Any]) -> str: return f"{registration['run_id']}:{registration['producer_id']}" +def _run(registration: Mapping[str, Any]) -> tuple[str, str]: + return str(registration["run_id"]), str(registration["producer_id"]) + + +def _credential_fingerprint(hf_token: str) -> str: + """Identify a credential in memory without keeping the token itself.""" + + return hashlib.sha256(hf_token.encode()).hexdigest() + + class CollectorDelivery: - """Authenticate queued runs and deliver idempotent event batches.""" + """Authenticate queued runs and deliver idempotent event batches. + + A service delivers its own producer's run. The spool is shared by every + build on the host, so it also holds other producers' runs; the service + adopts one only once that producer's service has gone (see ``leases``), one + adopter at a time. The collector gives a run to the Hugging Face user who + registers it first, and this service's credential is not the other run's, + so an answer about the credential (none, rejected, or not the run's owner) + makes only the service's own run local-only. Another producer's run goes + local-only only for an answer that does not depend on who asks. + """ def __init__( self, spool: EventSpool, + registration: Mapping[str, Any], *, development_collector_url: str | None = None, + leases: ProducerLeases | None = None, ) -> None: self.collector_url = ( _development_collector_url(development_collector_url) @@ -156,12 +185,36 @@ def __init__( else _collector_origin(PRODUCTION_COLLECTOR_URL) ) self.spool = spool + self._own_run = _run(registration) + self._leases = ( + leases if leases is not None else ProducerLeases.beside(spool.path) + ) + # Taken before the service is ready, so before its run has any event: + # no other service can find the run pending without a live lease. + self._own_lease = self._leases.hold( + *self._own_run, + timeout_seconds=OWN_LEASE_TIMEOUT_SECONDS, + ) self._session_token: tuple[str, float] | None = None self._registered: set[str] = set() self._warned_no_token = False self._warned_denied: set[str] = set() self._next_attempt_at = 0.0 self._retry_seconds = INITIAL_RETRY_SECONDS + self._next_lease_sweep_at = 0.0 + # Fingerprints of the credential last read and of those the collector + # refused; the token itself is never kept. + self._credential: str | None = None + self._rejected_credentials: set[str] = set() + # Other producers' runs this credential cannot deliver, by the + # credential that was refused; tried again once the credential changes. + self._passed_over: dict[str, str | None] = {} + + def close(self) -> None: + """Release this service's lease; its run, if still pending, is adoptable.""" + + if self._own_lease is not None: + self._own_lease.release() def flush_once(self) -> bool: """Attempt one delivery pass without waiting for retry deadlines.""" @@ -169,48 +222,119 @@ def flush_once(self) -> bool: if time.monotonic() < self._next_attempt_at: return False made_progress = False - for registration in self.spool.pending_runs(): - run_id = str(registration["run_id"]) - registration_key = _registration_key(registration) - token = self._collector_token(registration) - if token is None: - continue - if registration_key not in self._registered: - if not self._register(registration, token): - continue - events = self.spool.batch(run_id, str(registration["producer_id"])) - if not events: - continue - try: - status, _ = _http_post( - self.collector_url + RUN_EVENTS_PATH_TEMPLATE.format(run_id=run_id), - {"events": events}, - token, - ) - except (OSError, TimeoutError): - self._defer_retry() + pending = self.spool.pending_runs() + pending_keys = {_registration_key(registration) for registration in pending} + self._passed_over = { + key: credential + for key, credential in self._passed_over.items() + if key in pending_keys + } + self._sweep_leases_if_due(pending) + for registration in pending: + claim = self._claim(registration) + if claim is None: continue - if status == HTTPStatus.ACCEPTED: - self.spool.acknowledge([event["event_id"] for event in events]) - made_progress = True - self._retry_seconds = INITIAL_RETRY_SECONDS - elif status == HTTPStatus.UNAUTHORIZED: - self._session_token = None - elif status == HTTPStatus.FORBIDDEN: - self._make_local_only( - registration, - LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION, - ) - elif _permanent_client_error(status): - self._make_local_only( - registration, - LOCAL_ONLY_REJECTED_EVENTS, - message=REJECTED_EVENTS_MESSAGE.format(status=int(status)), - ) - else: - self._defer_retry() + with claim: + if self._deliver(registration): + made_progress = True return made_progress + def _claim( + self, + registration: Mapping[str, Any], + ) -> contextlib.AbstractContextManager[object] | None: + """Return a hold on a run this service may deliver now, else ``None``. + + The own run is always deliverable. Another producer's run is + deliverable only while this service holds that producer's lease, which + it can take only once the producing service has gone. A run with no + lease file (an older checkout's, or one whose free lease was swept) + is first left alone until it has been idle for ``ORPHAN_IDLE_SECONDS``. + """ + + run = _run(registration) + if run == self._own_run: + return contextlib.nullcontext() + key = _registration_key(registration) + if key in self._passed_over and self._passed_over[key] == self._credential: + return None + if not self._leases.supported: + return contextlib.nullcontext() if self._idle(run) else None + try: + try: + return self._leases.try_acquire(*run, create=False) + except FileNotFoundError: + if not self._idle(run): + return None + return self._leases.try_acquire(*run, create=True) + except OSError: + # A lease that cannot be tested may be live: leave the run be. + return None + + def _idle(self, run: tuple[str, str]) -> bool: + updated_at = self.spool.last_updated(*run) + if updated_at is None: + return False + return datetime.now(UTC) - updated_at >= timedelta(seconds=ORPHAN_IDLE_SECONDS) + + def _sweep_leases_if_due(self, pending: list[dict[str, Any]]) -> None: + now = time.monotonic() + if now < self._next_lease_sweep_at: + return + self._next_lease_sweep_at = now + LEASE_SWEEP_INTERVAL_SECONDS + keep = {_run(registration) for registration in pending} + keep.add(self._own_run) + with contextlib.suppress(OSError): + self._leases.sweep(keep) + + def _deliver(self, registration: Mapping[str, Any]) -> bool: + """Send one batch of a run's events; return whether any was delivered.""" + + run_id = str(registration["run_id"]) + registration_key = _registration_key(registration) + token = self._collector_token(registration) + if token is None: + return False + if registration_key not in self._registered: + if not self._register(registration, token): + return False + events = self.spool.batch(run_id, str(registration["producer_id"])) + if not events: + return False + try: + status, _ = _http_post( + self.collector_url + RUN_EVENTS_PATH_TEMPLATE.format(run_id=run_id), + {"events": events}, + token, + ) + except (OSError, TimeoutError): + self._defer_retry() + return False + if status == HTTPStatus.ACCEPTED: + self.spool.acknowledge([event["event_id"] for event in events]) + self._retry_seconds = INITIAL_RETRY_SECONDS + return True + if status == HTTPStatus.UNAUTHORIZED: + self._session_token = None + elif status == HTTPStatus.FORBIDDEN: + # The collector refuses the caller here (another Hugging Face user + # owns the run, say), so the answer depends on whose credential asked. + self._make_local_only( + registration, + 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 + def _collector_token(self, registration: Mapping[str, Any]) -> str | None: if ( self._session_token is not None @@ -219,10 +343,26 @@ def _collector_token(self, registration: Mapping[str, Any]) -> str | None: return self._session_token[0] hf_token = _huggingface_token() if not hf_token: - if not self._warned_no_token: + self._credential = None + if _run(registration) == self._own_run and not self._warned_no_token: print(NO_CREDENTIAL_MESSAGE, file=sys.stderr, flush=True) self._warned_no_token = True - self._make_local_only(registration, LOCAL_ONLY_MISSING_CREDENTIAL) + self._make_local_only( + registration, + LOCAL_ONLY_MISSING_CREDENTIAL, + caller_dependent=True, + ) + return None + credential = _credential_fingerprint(hf_token) + self._credential = credential + if credential in self._rejected_credentials: + # The collector has already refused this credential; asking again + # would get the same answer. + self._make_local_only( + registration, + LOCAL_ONLY_REJECTED_CREDENTIAL, + caller_dependent=True, + ) return None try: status, response = _http_post( @@ -242,7 +382,12 @@ def _collector_token(self, registration: Mapping[str, Any]) -> str | None: self._session_token = (token, time.monotonic() + expires_in) return token if status in {HTTPStatus.UNAUTHORIZED, HTTPStatus.FORBIDDEN}: - self._make_local_only(registration, LOCAL_ONLY_REJECTED_CREDENTIAL) + self._rejected_credentials.add(credential) + self._make_local_only( + registration, + LOCAL_ONLY_REJECTED_CREDENTIAL, + caller_dependent=True, + ) else: self._defer_retry() return None @@ -264,10 +409,21 @@ def _register(self, registration: Mapping[str, Any], token: str) -> bool: return True if status == HTTPStatus.UNAUTHORIZED: self._session_token = None - elif status in {HTTPStatus.FORBIDDEN, HTTPStatus.CONFLICT} or ( - _permanent_client_error(status) - ): - self._make_local_only(registration, LOCAL_ONLY_REJECTED_REGISTRATION) + elif status == HTTPStatus.FORBIDDEN: + # Another Hugging Face user registered the run first. + self._make_local_only( + registration, + LOCAL_ONLY_REJECTED_REGISTRATION, + caller_dependent=True, + ) + elif status == HTTPStatus.CONFLICT or _permanent_client_error(status): + # A conflict reaches only the run's owner, so it is about the + # registration itself, as is any other settled 4xx. + self._make_local_only( + registration, + LOCAL_ONLY_REJECTED_REGISTRATION, + caller_dependent=False, + ) else: self._defer_retry() return False @@ -277,13 +433,29 @@ def _make_local_only( registration: Mapping[str, Any], reason: str, *, + caller_dependent: bool, message: str = REJECTED_CREDENTIAL_MESSAGE, ) -> None: - run_id = str(registration["run_id"]) - producer_id = str(registration["producer_id"]) + """Stop uploading a run, unless the reason is only this service's. + + ``caller_dependent`` marks an outcome that turns on this service's own + credential: none, refused, or not the run's owner. That decides the + 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. + """ + + run_id, producer_id = _run(registration) registration_key = _registration_key(registration) + own = (run_id, producer_id) == self._own_run + if caller_dependent and not own: + if reason != LOCAL_ONLY_MISSING_CREDENTIAL: + self._passed_over[registration_key] = self._credential + return self.spool.make_local_only(run_id, producer_id, reason) - if reason != LOCAL_ONLY_MISSING_CREDENTIAL: + # 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(registration_key, 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 83163fe53c..362ef8ac23 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 @@ -60,6 +60,17 @@ DATABASE_TIMEOUT_SECONDS: Final = 5 PRUNE_INTERVAL_SECONDS: Final = 60.0 +LEASE_DIRECTORY_SUFFIX: Final = ".leases" +LEASE_FILE_SUFFIX: Final = ".lock" +LEASE_RETRY_SECONDS: Final = 0.01 +MAX_LEASE_OPEN_ATTEMPTS: Final = 8 +OWN_LEASE_TIMEOUT_SECONDS: Final = 1.0 +LEASE_SWEEP_INTERVAL_SECONDS: Final = 600.0 +#: A run with no lease file (its producer predates leases, or its free lease was +#: swept) counts as orphaned once it has gone this long without an update. A +#: live service appends a heartbeat every ``DEFAULT_HEARTBEAT_SECONDS``. +ORPHAN_IDLE_SECONDS: Final = 900.0 + DEFAULT_HEARTBEAT_SECONDS: Final = 60.0 DEFAULT_DRAIN_SECONDS: Final = 15.0 MINIMUM_HEARTBEAT_SECONDS: Final = 1.0 diff --git a/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/leases.py b/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/leases.py new file mode 100644 index 0000000000..60f76aeffe --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/leases.py @@ -0,0 +1,211 @@ +"""Advisory leases that tell a live producer's run from an orphaned one. + +Every emitter service on a host shares one spool, so the spool also holds other +builds' runs. A run belongs to the service that produced it while that service +lives. The collector gives a run to the Hugging Face user who registers it +first, so another service delivering a live run would register it under its own +credential, and the collector's answer to that credential says nothing about +the run's own. + +Each service therefore holds an exclusive ``flock`` on its producer's lease file +for its whole life. The kernel drops the lock when the process exits by any +route, SIGKILL included, so a lease that can be taken means the producing +service is gone and its run may be adopted. The lease files live beside the +spool, outside the database, so no spool migration is needed and checkouts that +predate leases can still share the spool. +""" + +from __future__ import annotations + +import hashlib +import os +import time +from collections.abc import Callable, Iterable +from pathlib import Path + +try: + import fcntl +except ImportError: # pragma: no cover - the service needs Unix sockets anyway + fcntl = None + +from microcosm.build.telemetry_emitter_service.constants import ( + LEASE_DIRECTORY_SUFFIX, + LEASE_FILE_SUFFIX, + LEASE_RETRY_SECONDS, + MAX_LEASE_OPEN_ATTEMPTS, +) + + +class ProducerLease: + """An exclusive lock on one producer's lease file, held until released.""" + + def __init__(self, path: Path, descriptor: int) -> None: + self.path = path + self._descriptor: int | None = descriptor + + @property + def held(self) -> bool: + return self._descriptor is not None + + def release(self, *, remove: bool = False) -> None: + """Drop the lock, first unlinking the file when ``remove`` is set. + + The file is unlinked while the lock is still held, so no other process + can be holding it then. Anyone who opened the old file meanwhile finds, + once it gets the lock, that the path no longer names that file, and + tries again. + """ + + if self._descriptor is None: + return + try: + if remove: + self.path.unlink(missing_ok=True) + finally: + os.close(self._descriptor) + self._descriptor = None + + def __enter__(self) -> ProducerLease: + return self + + def __exit__(self, *exc_info: object) -> None: + self.release() + + +class ProducerLeases: + """The lease files of every producer that shares one spool.""" + + def __init__(self, directory: Path | str) -> None: + self.directory = Path(directory) + + @classmethod + def beside(cls, spool_path: Path | str) -> ProducerLeases: + """Return the leases for the spool at ``spool_path``.""" + + spool_path = Path(spool_path) + return cls(spool_path.with_name(spool_path.name + LEASE_DIRECTORY_SUFFIX)) + + @property + def supported(self) -> bool: + """Whether this platform has advisory file locks.""" + + return fcntl is not None + + def path(self, run_id: str, producer_id: str) -> Path: + """Return the lease file of one producer. + + Identifiers come from the stored registration, so the file name is a + digest of them rather than the identifiers themselves. + """ + + digest = hashlib.sha256(f"{run_id}\0{producer_id}".encode()).hexdigest() + return self.directory / f"{digest}{LEASE_FILE_SUFFIX}" + + def exists(self, run_id: str, producer_id: str) -> bool: + return self.path(run_id, producer_id).exists() + + def try_acquire( + self, + run_id: str, + producer_id: str, + *, + create: bool, + ) -> ProducerLease | None: + """Take a producer's lease without waiting. + + Returns ``None`` while another process holds it. With ``create`` unset, + a missing lease file raises ``FileNotFoundError`` rather than creating + one, so the caller can tell a producer that never had a lease (an older + checkout) from one that has gone. Any other failure to test the lease + raises ``OSError``. + """ + + if fcntl is None: + raise OSError("advisory file locks are not available") + if create: + self.directory.mkdir(mode=0o700, parents=True, exist_ok=True) + return _lock(self.path(run_id, producer_id), create=create) + + def hold( + self, + run_id: str, + producer_id: str, + *, + timeout_seconds: float, + clock: Callable[[], float] = time.monotonic, + sleep: Callable[[float], None] = time.sleep, + ) -> ProducerLease | None: + """Take this service's own lease, waiting out brief holders. + + Only a sweeper or an adopter holds another producer's lease, and only + for a moment, so a short wait suffices. Returns ``None`` if the lease is + still taken at the deadline or cannot be taken at all; the service then + runs without one, and other services judge its run by recent activity. + """ + + deadline = clock() + timeout_seconds + while True: + try: + lease = self.try_acquire(run_id, producer_id, create=True) + except OSError: + return None + if lease is not None or clock() >= deadline: + return lease + sleep(LEASE_RETRY_SECONDS) + + def sweep(self, keep: Iterable[tuple[str, str]]) -> int: + """Remove free lease files of producers that have nothing pending. + + A held lease belongs to a live service and is never removed, nor is the + lease of a run in ``keep`` (the pending runs), so an orphan stays + adoptable at once. Returns the number of files removed. + """ + + if fcntl is None or not self.directory.is_dir(): + return 0 + kept = {self.path(run_id, producer_id).name for run_id, producer_id in keep} + removed = 0 + for path in self.directory.glob(f"*{LEASE_FILE_SUFFIX}"): + if path.name in kept: + continue + try: + lease = _lock(path, create=False) + except OSError: + continue + if lease is not None: + lease.release(remove=True) + removed += 1 + return removed + + +def _lock(path: Path, *, create: bool) -> ProducerLease | None: + flags = os.O_RDWR | getattr(os, "O_CLOEXEC", 0) + if create: + flags |= os.O_CREAT + for _ in range(MAX_LEASE_OPEN_ATTEMPTS): + descriptor = os.open(path, flags, 0o600) + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + os.close(descriptor) + return None + except BaseException: + os.close(descriptor) + raise + if _names_open_file(path, descriptor): + return ProducerLease(path, descriptor) + # Another process removed this file between our open and our lock. + os.close(descriptor) + if not create and not path.exists(): + raise FileNotFoundError(path) + # The path keeps being replaced under us: someone is taking it. + return None + + +def _names_open_file(path: Path, descriptor: int) -> bool: + try: + on_disk = path.stat() + except FileNotFoundError: + return False + opened = os.fstat(descriptor) + return (on_disk.st_dev, on_disk.st_ino) == (opened.st_dev, opened.st_ino) diff --git a/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/main.py b/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/main.py index 2872bf91a8..9216f09569 100644 --- a/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/main.py +++ b/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/main.py @@ -38,16 +38,23 @@ def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) registration = json.loads(args.registration_json) spool = EventSpool(args.spool) + # The delivery takes this producer's lease, which must happen before the + # service is ready: from then on the run can have events. + delivery = CollectorDelivery( + spool, + registration, + development_collector_url=args.development_collector_url, + ) service = EmitterService( socket_path=args.socket, registration=registration, spool=spool, - delivery=CollectorDelivery( - spool, - development_collector_url=args.development_collector_url, - ), + delivery=delivery, sampler=ProcessTreeSampler(args.parent_pid), heartbeat_seconds=args.heartbeat_seconds, ) - service.run() + try: + service.run() + finally: + delivery.close() return 0 diff --git a/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/runtime.py b/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/runtime.py index 0bb1d9c2a4..e0eb42f080 100644 --- a/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/runtime.py +++ b/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/runtime.py @@ -211,8 +211,15 @@ def _worker(self) -> None: self._drain() def _drain(self) -> None: + # Only this build's run holds the service up: other builds' runs are + # their own services' to deliver, or are adopted while this one runs. + run_id = str(self.registration["run_id"]) + producer_id = str(self.registration["producer_id"]) deadline = time.monotonic() + self.drain_seconds - while self.spool.has_deliverable() and time.monotonic() < deadline: + while ( + self.spool.run_has_deliverable(run_id, producer_id) + and time.monotonic() < deadline + ): if not self.delivery.flush_once(): remaining = deadline - time.monotonic() if remaining > 0: diff --git a/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/spool.py b/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/spool.py index 1ef43c5269..f852f25a83 100644 --- a/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/spool.py +++ b/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/spool.py @@ -170,6 +170,39 @@ def has_deliverable(self) -> bool: with self._lock, self._session_factory() as session: return session.scalar(statement) is not None + def run_has_deliverable(self, run_id: str, producer_id: str) -> bool: + """Return whether one producer still has events eligible for delivery.""" + + statement = ( + select(TelemetryEventRecord.event_id) + .join(TelemetryEventRecord.run) + .where( + TelemetryRunRecord.run_id == run_id, + TelemetryRunRecord.producer_id == producer_id, + TelemetryRunRecord.upload_state == UPLOAD_STATE_PENDING, + ) + .limit(1) + ) + with self._lock, self._session_factory() as session: + return session.scalar(statement) is not None + + def last_updated(self, run_id: str, producer_id: str) -> datetime | None: + """Return when one producer's run last changed, or ``None`` if unknown.""" + + statement = select(TelemetryRunRecord.updated_at).where( + TelemetryRunRecord.run_id == run_id, + TelemetryRunRecord.producer_id == producer_id, + ) + with self._lock, self._session_factory() as session: + updated_at = session.scalar(statement) + if updated_at is None: + return None + try: + parsed = datetime.fromisoformat(updated_at) + except ValueError: + return None + return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC) + def batch( self, run_id: str, 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 cee28f4349..af17b364fa 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 @@ -110,7 +110,9 @@ def test_production_collector_cannot_be_replaced_by_environment( "https://untrusted.example", ) - delivery = CollectorDelivery(EventSpool(tmp_path / "events.sqlite3")) + delivery = CollectorDelivery( + EventSpool(tmp_path / "events.sqlite3"), _registration() + ) assert delivery.collector_url == PRODUCTION_COLLECTOR_URL @@ -386,6 +388,7 @@ def fake_post(url, payload, token, *, timeout=5.0): delivery = CollectorDelivery( spool, + registration, development_collector_url="http://127.0.0.1:8080", ) assert delivery.flush_once() @@ -416,6 +419,7 @@ def reject(url, payload, token, *, timeout=5.0): monkeypatch.setattr(collector_module, "_http_post", reject) delivery = CollectorDelivery( spool, + registration, development_collector_url="http://127.0.0.1:8080", ) @@ -451,7 +455,7 @@ def fake_post(url, payload, token, *, timeout=5.0): monkeypatch.setattr(collector_module, "_http_post", fake_post) delivery = CollectorDelivery( - spool, development_collector_url="http://127.0.0.1:8080" + spool, registration, development_collector_url="http://127.0.0.1:8080" ) assert not delivery.flush_once() assert spool.has_pending() @@ -480,7 +484,7 @@ def test_identity_provider_outage_keeps_events_eligible_for_retry( lambda *args, **kwargs: (503, {"detail": "temporarily unavailable"}), ) - assert not CollectorDelivery(spool).flush_once() + assert not CollectorDelivery(spool, registration).flush_once() assert spool.pending_runs() == [registration] @@ -501,7 +505,7 @@ def test_missing_credential_never_contacts_collector_and_stays_local_only( ), ) - delivery = CollectorDelivery(spool) + delivery = CollectorDelivery(spool, registration) assert not delivery.flush_once() assert spool.pending_runs() == [] @@ -895,7 +899,7 @@ def sleep(seconds: float) -> None: service = EmitterService( socket_path=tmp_path / "unused.sock", registration=_registration(), - spool=SimpleNamespace(has_deliverable=lambda: True), + spool=SimpleNamespace(run_has_deliverable=lambda run_id, producer_id: True), delivery=SimpleNamespace(flush_once=flush_once), sampler=_FakeSampler(), heartbeat_seconds=60, 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 new file mode 100644 index 0000000000..aa88d3c6a3 --- /dev/null +++ b/packages/microcosm-build/tests/engine_free/shared/test_telemetry_foreign_runs.py @@ -0,0 +1,820 @@ +"""A telemetry service decides only its own run's fate from its own credential. + +Every emitter service on a host shares one spool, which therefore also holds +other builds' runs. These tests run several producers against one spool and a +fake collector that follows the hosted collector's authorization model +(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; an +unknown run's events get 404, and a batch the collector cannot accept 422. + +Invariants (each checked below, the first three for any interleaving): + +- 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. +- While a producer's service is alive, no other service sends any request for + that producer's run. +- A run goes local-only only right after a collector answer (or, for a missing + credential, the absence of one) that justifies it, and a refused credential + is exchanged once. +- Once every service has gone, a run left pending is delivered in full by a + service whose credential owns it. +""" + +from __future__ import annotations + +import json +import signal +import sqlite3 +import subprocess +import sys +import textwrap +import threading +import time +from contextlib import closing +from dataclasses import dataclass, field +from datetime import UTC, datetime, timedelta +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from types import SimpleNamespace +from typing import Any +from unittest import mock +from urllib.parse import urlsplit + +import pytest +from hypothesis import HealthCheck, event, given, settings +from hypothesis import strategies as st + +from microcosm.build.telemetry_emitter import LocalTelemetryEmitter, TelemetryRun +from microcosm.build.telemetry_emitter_service import ( + CollectorDelivery, + EmitterService, + EventSpool, +) +from microcosm.build.telemetry_emitter_service import collector as collector_module +from microcosm.build.telemetry_emitter_service import leases as leases_module +from microcosm.build.telemetry_emitter_service.constants import ( + LOCAL_ONLY_MISSING_CREDENTIAL, + LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION, + LOCAL_ONLY_REJECTED_CREDENTIAL, + LOCAL_ONLY_REJECTED_EVENTS, + LOCAL_ONLY_REJECTED_REGISTRATION, + ORPHAN_IDLE_SECONDS, + RUN_REGISTRATION_PATH, + TOKEN_EXCHANGE_PATH, +) +from microcosm.build.telemetry_emitter_service.leases import ProducerLeases +from microcosm.build.telemetry_protocol import BUILD_COMPLETED_MESSAGE + +MEMBER = "hf-member:" +OUTSIDER = "hf-outsider" +EXPIRED = "hf-expired" +CREDENTIALS = (None, OUTSIDER, EXPIRED, MEMBER + "alice", MEMBER + "bob") +DEVELOPMENT_COLLECTOR = "http://127.0.0.1:8080" +EVENT = { + "timestamp": "2026-10-09T00:00:00+00:00", + "event_type": "stage", + "stage_id": "compile", + "status": "started", + "message": None, + "details": {}, +} + +Run = tuple[str, str] + + +def _registration(run_id: str, producer_id: str) -> dict[str, Any]: + return TelemetryRun( + run_id=run_id, + country_code="US", + pipeline="foreign-runs", + producer_id=producer_id, + ).as_registration() + + +class FakeCollector: + """The hosted collector's answers, as its authorization model gives them.""" + + def __init__(self, rejected_run_ids: set[str] | None = None) -> None: + self.rejected_run_ids = set(rejected_run_ids or ()) + self.owners: dict[str, str] = {} + self.metadata: dict[str, dict[str, Any]] = {} + self.producers: set[Run] = set() + self.accepted: dict[Run, set[str]] = {} + # (caller's run, kind, run asked about or None, status) + self.requests: list[tuple[Run, str, Run | None, int]] = [] + # (caller's run, Hugging Face token) for every refused exchange + self.refused: list[tuple[Run, str]] = [] + + def post(self, caller: Run, url: str, payload, token: str) -> tuple[int, dict]: + path = urlsplit(url).path + if path == TOKEN_EXCHANGE_PATH: + if not token.startswith(MEMBER): + self.refused.append((caller, token)) + if token.startswith(MEMBER): + user = token.removeprefix(MEMBER) + return self._answer( + caller, + "exchange", + None, + 200, + {"access_token": f"session:{user}", "expires_in": 3600}, + ) + return self._answer( + caller, "exchange", None, 403 if token == OUTSIDER else 401, {} + ) + user = token.removeprefix("session:") + if path == RUN_REGISTRATION_PATH: + run = (payload["run_id"], payload["producer_id"]) + 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"} + if self.metadata.setdefault(run[0], metadata) != metadata: + return self._answer(caller, "register", run, 409, {}) + self.producers.add(run) + return self._answer(caller, "register", run, 201, {"registered": True}) + run_id = path.split("/")[3] + events = payload["events"] + run = (run_id, events[0]["producer_id"]) + if run_id not in self.owners: + 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, {}) + if run_id in self.rejected_run_ids: + return self._answer(caller, "events", run, 422, {}) + self.accepted.setdefault(run, set()).update(e["event_id"] for e in events) + return self._answer(caller, "events", run, 202, {"accepted": len(events)}) + + def _answer(self, caller, kind, run, status, body): + self.requests.append((caller, kind, run, status)) + return status, body + + def asked(self, caller: Run, kind: str, run: Run | None = None) -> list[int]: + return [ + status + for who, what, about, status in self.requests + if who == caller and what == kind and (run is None or about == run) + ] + + +@dataclass +class Producer: + """One build's emitter service: its own spool connection, lease and login.""" + + registration: dict[str, Any] + credential: str | None + spool: EventSpool + delivery: CollectorDelivery + alive: bool = True + appended: list[str] = field(default_factory=list) + + @property + def run(self) -> Run: + return self.registration["run_id"], self.registration["producer_id"] + + +class Host: + """Several builds on one machine, sharing one spool and one collector.""" + + def __init__(self, directory: Path, collector: FakeCollector) -> None: + self.spool_path = directory / "events.sqlite3" + self.collector = collector + self.producers: list[Producer] = [] + # (caller's run, run made local-only, reason, caller's login then) + self.local_only: list[tuple[Run, Run, str, str | None]] = [] + # Requests a service sent about another producer's run while it lived. + self.intrusions: list[tuple[Run, str, Run]] = [] + + def start(self, credential: str | None, run_id: str | None = None) -> Producer: + index = len(self.producers) + registration = _registration(run_id or f"run-{index}", f"producer-{index}") + spool = EventSpool(self.spool_path) + # As in the service's main(): the lease is taken before registration. + delivery = CollectorDelivery( + spool, registration, development_collector_url=DEVELOPMENT_COLLECTOR + ) + spool.register(registration) + producer = Producer(registration, credential, spool, delivery) + make_local_only = spool.make_local_only + + def recorded(run_id: str, producer_id: str, reason: str) -> None: + self.local_only.append( + (producer.run, (run_id, producer_id), reason, producer.credential) + ) + make_local_only(run_id, producer_id, reason) + + spool.make_local_only = recorded + self.producers.append(producer) + return producer + + def emit(self, producer: Producer, count: int = 1) -> None: + assert producer.alive + for _ in range(count): + event = producer.spool.append(producer.registration, EVENT) + producer.appended.append(event["event_id"]) + + def flush(self, producer: Producer) -> bool: + assert producer.alive + + def post(url, payload, token, *, timeout=5.0): + status, body = self.collector.post(producer.run, url, payload, token) + about = self.collector.requests[-1][2] + owner = self.producer_of(about) if about is not None else None + if owner is not None and owner is not producer and owner.alive: + self.intrusions.append((producer.run, url, about)) + return status, body + + with ( + mock.patch.object( + collector_module, "_huggingface_token", lambda: producer.credential + ), + mock.patch.object(collector_module, "_http_post", post), + ): + return producer.delivery.flush_once() + + def kill(self, producer: Producer) -> None: + producer.alive = False + producer.delivery.close() + + def producer_of(self, run: Run) -> Producer | None: + return next((p for p in self.producers if p.run == run), None) + + def states(self) -> dict[Run, tuple[str, str | None]]: + with closing(sqlite3.connect(self.spool_path)) as connection: + return { + (run_id, producer_id): (state, reason) + for run_id, producer_id, state, reason in connection.execute( + "SELECT run_id, producer_id, upload_state, local_only_reason " + "FROM telemetry_runs" + ) + } + + def queued(self, run: Run) -> int: + with closing(sqlite3.connect(self.spool_path)) as connection: + return connection.execute( + "SELECT COUNT(*) FROM telemetry_events " + "WHERE run_id = ? AND producer_id = ?", + run, + ).fetchone()[0] + + def close(self) -> None: + for producer in self.producers: + producer.delivery.close() + producer.spool._engine.dispose() + + +@pytest.fixture +def host(tmp_path): + host = Host(tmp_path, FakeCollector()) + yield host + host.close() + + +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 + for caller, run, reason, credential in host.local_only: + if caller != run: + # Another producer's run: only the collector's answer about the + # run's data, never one about this service's credential. + assert reason == LOCAL_ONLY_REJECTED_EVENTS, (caller, run, reason) + assert run[0] in rejected + assert 422 in host.collector.asked(caller, "events", run) + continue + if reason == LOCAL_ONLY_MISSING_CREDENTIAL: + assert credential is None + elif reason == LOCAL_ONLY_REJECTED_CREDENTIAL: + assert set(host.collector.asked(caller, "exchange")) & {401, 403} + elif reason == LOCAL_ONLY_REJECTED_REGISTRATION: + assert set(host.collector.asked(caller, "register", run)) & {403, 409} + elif reason == LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION: + assert 403 in host.collector.asked(caller, "events", run) + else: + assert reason == LOCAL_ONLY_REJECTED_EVENTS + assert set(host.collector.asked(caller, "events", run)) & {404, 422} + # A refused credential is exchanged once per service, not on every pass. + refused = host.collector.refused + assert len(set(refused)) == len(refused), refused + + +def test_a_service_without_a_credential_leaves_a_live_build_run_pending(host): + """The 2026-10-09 e7 reproduction: build B's worker had no credential and + made build A's run local-only.""" + + build_a = host.start(MEMBER + "alice") + build_b = host.start(None) + host.emit(build_a) + host.emit(build_b) + + host.flush(build_b) + + assert host.states()[build_a.run] == ("pending", None) + assert host.states()[build_b.run] == ("local_only", LOCAL_ONLY_MISSING_CREDENTIAL) + assert host.collector.requests == [] + host.flush(build_a) + assert host.collector.accepted[build_a.run] == set(build_a.appended) + _assert_own_credential_only(host) + + +@pytest.mark.parametrize("credential", CREDENTIALS) +def test_a_live_build_run_gets_no_request_from_another_service(host, credential): + build_a = host.start(MEMBER + "alice") + build_b = host.start(credential) + host.emit(build_a, 3) + host.emit(build_b) + + for _ in range(3): + host.flush(build_b) + + assert all(about != build_a.run for _, _, about, _ in host.collector.requests) + assert host.states()[build_a.run] == ("pending", None) + assert host.queued(build_a.run) == 3 + _assert_own_credential_only(host) + + +@pytest.mark.parametrize( + ("credential", "own_reason", "exchanges"), + [ + (None, LOCAL_ONLY_MISSING_CREDENTIAL, 0), + (OUTSIDER, LOCAL_ONLY_REJECTED_CREDENTIAL, 1), + (EXPIRED, LOCAL_ONLY_REJECTED_CREDENTIAL, 1), + ], +) +def test_an_orphan_waits_for_a_credential_that_can_deliver_it( + host, capsys, credential, own_reason, exchanges +): + build_a = host.start(MEMBER + "alice") + host.emit(build_a, 3) + host.kill(build_a) + build_b = host.start(credential) + host.emit(build_b) + + for _ in range(4): + host.flush(build_b) + + assert host.states()[build_a.run] == ("pending", None) + assert host.states()[build_b.run] == ("local_only", own_reason) + assert len(host.collector.asked(build_b.run, "exchange")) == exchanges + assert capsys.readouterr().err.count("local-only") == 1 + + rescuer = host.start(MEMBER + "alice") + host.flush(rescuer) + assert host.collector.accepted[build_a.run] == set(build_a.appended) + assert host.queued(build_a.run) == 0 + assert host.states()[build_a.run] == ("pending", None) + _assert_own_credential_only(host) + + +def test_an_orphan_owned_by_another_user_is_passed_over_not_made_local_only(host): + build_a = host.start(MEMBER + "alice") + host.emit(build_a, 2) + host.flush(build_a) + host.emit(build_a, 2) + host.kill(build_a) + build_b = host.start(MEMBER + "bob") + host.emit(build_b) + + for _ in range(3): + host.flush(build_b) + + # Bob may not register Alice's run; asking once settles it for Bob. + assert host.collector.asked(build_b.run, "register", build_a.run) == [403] + assert host.states()[build_a.run] == ("pending", None) + assert host.queued(build_a.run) == 2 + assert host.collector.accepted[build_b.run] == set(build_b.appended) + + rescuer = host.start(MEMBER + "alice") + host.flush(rescuer) + assert host.collector.accepted[build_a.run] == set(build_a.appended) + _assert_own_credential_only(host) + + +@pytest.mark.parametrize("answer", ["unacceptable_events", "registration_conflict"]) +def test_an_orphan_the_collector_refuses_goes_local_only_without_a_warning_here( + host, capsys, answer +): + build_a = host.start(MEMBER + "alice") + host.emit(build_a) + host.kill(build_a) + if answer == "unacceptable_events": + host.collector.rejected_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. + host.collector.owners[build_a.run[0]] = "alice" + host.collector.metadata[build_a.run[0]] = {"pipeline": "something-else"} + reason = LOCAL_ONLY_REJECTED_REGISTRATION + build_b = host.start(MEMBER + "alice") + host.emit(build_b) + + host.flush(build_b) + + 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 + + +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") + host.emit(build_a) + host.emit(build_b) + + host.flush(build_a) + host.flush(build_b) + + assert host.states()[build_a.run] == ("pending", None) + assert host.states()[build_b.run] == ( + "local_only", + LOCAL_ONLY_REJECTED_REGISTRATION, + ) + assert host.collector.accepted[build_a.run] == set(build_a.appended) + _assert_own_credential_only(host) + + +def test_a_run_without_a_lease_is_adopted_only_once_idle(host): + """A run from a checkout that predates leases has no lease file, so only + its recent activity says whether its service may still be running.""" + + legacy = _registration("run-legacy", "producer-legacy") + older_checkout = EventSpool(host.spool_path) + older_checkout.register(legacy) + appended = {older_checkout.append(legacy, EVENT)["event_id"]} + older_checkout._engine.dispose() + legacy_run = (legacy["run_id"], legacy["producer_id"]) + adopter = host.start(MEMBER + "alice") + + host.flush(adopter) + assert all(about != legacy_run for _, _, about, _ in host.collector.requests) + + idle_since = datetime.now(UTC) - timedelta(seconds=ORPHAN_IDLE_SECONDS + 60) + with closing(sqlite3.connect(host.spool_path)) as connection: + connection.execute( + "UPDATE telemetry_runs SET updated_at = ? WHERE run_id = ?", + (idle_since.isoformat(), legacy["run_id"]), + ) + connection.commit() + host.flush(adopter) + + assert host.collector.accepted[legacy_run] == appended + leases = ProducerLeases.beside(host.spool_path) + lease = leases.try_acquire(*legacy_run, create=False) + assert lease is not None, "the adopter must hold the lease only while delivering" + lease.release() + + +def test_shutdown_drain_waits_only_for_its_own_run(tmp_path): + spool = EventSpool(tmp_path / "events.sqlite3") + own = _registration("run-own", "producer-own") + other = _registration("run-other", "producer-other") + spool.register(own) + spool.register(other) + spool.append(other, EVENT) + flushes: list[str] = [] + service = EmitterService( + socket_path=tmp_path / "unused.sock", + registration=own, + spool=spool, + delivery=SimpleNamespace(flush_once=lambda: flushes.append("other") or False), + sampler=SimpleNamespace(sample=lambda: None, parent_alive=lambda: True), + heartbeat_seconds=60, + drain_seconds=30, + ) + + started = time.monotonic() + service._drain() + assert flushes == [] + assert time.monotonic() - started < 5 + + spool.append(own, EVENT) + + def deliver_own() -> bool: + spool.acknowledge( + [e["event_id"] for e in spool.batch("run-own", "producer-own")] + ) + flushes.append("own") + return True + + service.delivery = SimpleNamespace(flush_once=deliver_own) + service._drain() + assert flushes == ["own"] + assert spool.has_deliverable() # the other build's run, left to its service + spool._engine.dispose() + + +def test_a_lease_is_exclusive_until_released(tmp_path): + leases = ProducerLeases(tmp_path / "leases") + held = leases.hold("run-a", "producer-a", timeout_seconds=1) + assert held is not None + + assert leases.try_acquire("run-a", "producer-a", create=False) is None + assert leases.hold("run-a", "producer-a", timeout_seconds=0.05) is None + with pytest.raises(FileNotFoundError): + leases.try_acquire("run-b", "producer-b", create=False) + + held.release() + taken = leases.try_acquire("run-a", "producer-a", create=False) + assert taken is not None + taken.release() + + +def test_sweep_removes_only_free_leases_of_runs_with_nothing_pending(tmp_path): + leases = ProducerLeases(tmp_path / "leases") + live = leases.hold("run-live", "p", timeout_seconds=1) + for run in (("run-orphan", "p"), ("run-done", "p")): + leases.try_acquire(*run, create=True).release() + + assert leases.sweep(keep={("run-orphan", "p")}) == 1 + + assert leases.exists("run-live", "p") + assert leases.exists("run-orphan", "p") + assert not leases.exists("run-done", "p") + live.release() + + +def test_a_lease_file_replaced_between_open_and_lock_is_not_trusted( + tmp_path, monkeypatch +): + leases = ProducerLeases(tmp_path / "leases") + path = leases.path("run-a", "producer-a") + real = leases_module.fcntl + calls: list[int] = [] + + def flock(descriptor: int, operation: int) -> None: + calls.append(descriptor) + if len(calls) == 1: + path.unlink() # a sweeper removes the file between open and lock + real.flock(descriptor, operation) + + monkeypatch.setattr( + leases_module, + "fcntl", + SimpleNamespace(flock=flock, LOCK_EX=real.LOCK_EX, LOCK_NB=real.LOCK_NB), + ) + lease = leases.try_acquire("run-a", "producer-a", create=True) + + assert lease is not None + assert len(calls) == 2 + # The lock is on the file now at the path, so nobody else can take it. + monkeypatch.setattr(leases_module, "fcntl", real) + assert leases.try_acquire("run-a", "producer-a", create=False) is None + lease.release() + + # Without create, a file removed under us reads as no lease at all. + calls.clear() + monkeypatch.setattr( + leases_module, + "fcntl", + SimpleNamespace(flock=flock, LOCK_EX=real.LOCK_EX, LOCK_NB=real.LOCK_NB), + ) + with pytest.raises(FileNotFoundError): + leases.try_acquire("run-a", "producer-a", create=False) + + +def test_a_killed_service_releases_its_lease(tmp_path): + leases = ProducerLeases(tmp_path / "leases") + holder = textwrap.dedent( + """ + import sys, time + from microcosm.build.telemetry_emitter_service.leases import ProducerLeases + lease = ProducerLeases(sys.argv[1]).hold( + "run-a", "producer-a", timeout_seconds=5 + ) + print("held" if lease is not None else "missing", flush=True) + time.sleep(600) + """ + ) + process = subprocess.Popen( + [sys.executable, "-c", holder, str(leases.directory)], + stdout=subprocess.PIPE, + text=True, + ) + try: + assert process.stdout.readline().strip() == "held" + assert leases.try_acquire("run-a", "producer-a", create=False) is None + + process.send_signal(signal.SIGKILL) + process.wait(timeout=30) + + lease = leases.try_acquire("run-a", "producer-a", create=False) + assert lease is not None + lease.release() + finally: + if process.poll() is None: + process.kill() + process.wait(timeout=30) + process.stdout.close() + + +def test_two_real_services_on_one_spool_leave_each_others_runs_alone( + tmp_path, monkeypatch, real_local_telemetry +): + """Build A has a credential, build B has none, and both services share one + spool. A's token exchange is held until B's worker has passed over A's + queued events several times; A's run must still arrive in full.""" + + release = threading.Event() + received: dict[str, list[dict[str, Any]]] = {} + lock = threading.Lock() + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + length = int(self.headers.get("Content-Length", "0")) + payload = json.loads(self.rfile.read(length)) + if self.path.endswith("/exchange"): + release.wait(timeout=30) + status = 200 + body = {"access_token": "collector-token", "expires_in": 3600} + elif self.path == "/v1/runs": + status, body = 201, {"registered": True} + else: + with lock: + received.setdefault(payload["events"][0]["run_id"], []).extend( + payload["events"] + ) + status = 202 + body = {"accepted": len(payload["events"]), "duplicates": 0} + encoded = json.dumps(body).encode() + try: + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(encoded))) + self.end_headers() + self.wfile.write(encoded) + except OSError: + pass # the service gave up waiting and retries + + def log_message(self, format, *args): + return + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + server.daemon_threads = True + server_thread = threading.Thread(target=server.serve_forever, daemon=True) + server_thread.start() + spool_path = tmp_path / "shared.sqlite3" + common = { + "country_code": "US", + "pipeline": "test-pipeline", + "development_collector_url": f"http://127.0.0.1:{server.server_port}", + "spool_path": spool_path, + "heartbeat_seconds": 60, + "startup_timeout_seconds": 60, + } + builds: list[LocalTelemetryEmitter] = [] + try: + monkeypatch.setenv("HF_TOKEN", "hf-build-a-token") + builds.append(LocalTelemetryEmitter.start(run_id="build-a", **common)) + monkeypatch.delenv("HF_TOKEN") + builds.append(LocalTelemetryEmitter.start(run_id="build-b", **common)) + build_a, build_b = builds + assert build_a.available and build_b.available + build_a.stage("compile", message="Started.") + build_b.stage("compile", message="Started.") + time.sleep(3.5) # B's worker passes over A's queued run every second + release.set() + build_a.complete() + build_b.complete() + for build in builds: + build._process.wait(timeout=60) + finally: + release.set() + for build in builds: + if build._process is not None and build._process.poll() is None: + build._process.kill() + build._process.wait(timeout=30) + server.shutdown() + server_thread.join(timeout=5) + server.server_close() + + events = sorted(received["build-a"], key=lambda event: event["sequence"]) + assert [e["sequence"] for e in events] == list(range(1, len(events) + 1)) + assert events[-1]["message"] == BUILD_COMPLETED_MESSAGE + assert "build-b" not in received + with closing(sqlite3.connect(spool_path)) as connection: + states = dict( + connection.execute( + "SELECT run_id, COALESCE(local_only_reason, upload_state) " + "FROM telemetry_runs" + ) + ) + assert states == { + "build-a": "pending", + "build-b": LOCAL_ONLY_MISSING_CREDENTIAL, + } + + +def _label_paths(host: Host) -> None: + """Report which delivery paths a generated scenario reached.""" + + for caller, kind, about, status in host.collector.requests: + if about is not None and about != caller: + event(f"adopted orphan: {kind} {status}") + for caller, run, reason, _ in host.local_only: + event(f"{'own' if caller == run else 'adopted'} run local-only: {reason}") + + +PRODUCER = st.integers(0, 7) +# A new build: its login, and whether it reuses an earlier build's run id. +BUILD = st.tuples(st.sampled_from(CREDENTIALS), st.integers(0, 3)) +OPERATIONS = st.lists( + st.one_of( + st.tuples(st.just("start"), BUILD), + st.tuples(st.just("emit"), PRODUCER, st.integers(1, 3)), + # Flushes are what decide runs' fates, so they come up most often. + st.tuples(st.just("flush"), PRODUCER), + st.tuples(st.just("flush"), PRODUCER), + st.tuples(st.just("flush"), PRODUCER), + st.tuples(st.just("kill"), PRODUCER), + st.tuples(st.just("login"), PRODUCER, st.sampled_from(CREDENTIALS)), + ), + max_size=30, +) + + +def _start_build(host: Host, build: tuple[str | None, int]) -> Producer: + credential, reuse = build + shared = ( + host.producers[(reuse - 1) % len(host.producers)].run[0] + if reuse and host.producers + else None + ) + return host.start(credential, run_id=shared) + + +@settings( + max_examples=150, + deadline=None, + suppress_health_check=[HealthCheck.function_scoped_fixture, HealthCheck.too_slow], +) +@given( + builds=st.lists(BUILD, min_size=2, max_size=4), + operations=OPERATIONS, + rejected=st.sets(st.integers(0, 7), max_size=2), +) +def test_no_service_decides_another_producers_run_from_its_own_credential( + tmp_path_factory, builds, operations, rejected +): + host = Host( + tmp_path_factory.mktemp("foreign"), + FakeCollector({f"run-{index}" for index in rejected}), + ) + try: + # Builds already running, each with an event queued. + for build in builds: + host.emit(_start_build(host, build)) + for operation in operations: + kind = operation[0] + if kind == "start": + _start_build(host, operation[1]) + continue + live = [producer for producer in host.producers if producer.alive] + if not live: + continue + producer = live[operation[1] % len(live)] + if kind == "emit": + host.emit(producer, operation[2]) + elif kind == "flush": + host.flush(producer) + elif kind == "kill": + host.kill(producer) + else: + producer.credential = operation[2] + _assert_own_credential_only(host) + _label_paths(host) + + # Every service goes. A run still pending is an orphan, and a service + # whose login owns it delivers it in full. + for producer in host.producers: + if producer.alive: + host.kill(producer) + rescuers = [host.start(MEMBER + user) for user in ("alice", "bob")] + before = host.states() + for _ in range(3): + for rescuer in rescuers: + host.flush(rescuer) + _assert_own_credential_only(host) + after = host.states() + for producer in host.producers: + if producer in rescuers or before[producer.run][0] != "pending": + continue + if producer.run[0] in host.collector.rejected_run_ids and ( + producer.appended + ): + if host.queued(producer.run): + assert after[producer.run] == ( + "local_only", + LOCAL_ONLY_REJECTED_EVENTS, + ) + continue + assert after[producer.run] == ("pending", None), producer.run + assert host.queued(producer.run) == 0, producer.run + assert host.collector.accepted.get(producer.run, set()) >= set( + producer.appended + ) + finally: + host.close() From 662fd4acb3742b07e9f178545f54b9c92a0d55a2 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Fri, 9 Oct 2026 14:37:56 -0400 Subject: [PATCH 2/4] Re-check a run's state after claiming it, before sending anything A delivery pass lists the pending runs once, then waits on the collector for seconds. Meanwhile another build's service can make its own run local-only and exit, freeing its lease, and the pass then adopted that run and uploaded it under its own login. The two-service test caught it. The run's state is now read again while it is held. Tests: a deterministic case for that interleaving; the property test can now run one service while another's request is in flight, and checks that nothing about a run reaches the collector after it goes local-only (the shrunk counterexample is pinned as an example). The two-service test waits for build B's first pass instead of sleeping, and fails on origin/main with build A's run local-only. Co-Authored-By: Claude Opus 5.5 --- .../telemetry_emitter_service/collector.py | 6 + .../shared/test_telemetry_foreign_runs.py | 115 +++++++++++++++--- 2 files changed, 106 insertions(+), 15 deletions(-) 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 6e7b7b22d5..12f1cf9bd0 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 @@ -235,6 +235,12 @@ def flush_once(self) -> bool: if claim is None: continue with claim: + # The list was read before this pass's requests, which can + # take seconds. Meanwhile a run's own service may have made it + # local-only and exited, freeing its lease. Only now, holding + # the run, is its state settled. + if not self.spool.run_has_deliverable(*_run(registration)): + continue if self._deliver(registration): made_progress = True return made_progress 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 aa88d3c6a3..0f44953245 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 @@ -8,7 +8,9 @@ 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. -Invariants (each checked below, the first three for any interleaving): +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 +another's collector request is in flight): - 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 @@ -18,6 +20,8 @@ - A run goes local-only only right after a collector answer (or, for a missing credential, the absence of one) that justifies it, and a refused credential is exchanged once. +- Local-only is final: no request about a run reaches the collector after it + 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. """ @@ -32,6 +36,7 @@ import textwrap import threading import time +from collections.abc import Callable from contextlib import closing from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta @@ -43,7 +48,7 @@ from urllib.parse import urlsplit import pytest -from hypothesis import HealthCheck, event, given, settings +from hypothesis import HealthCheck, event, example, given, settings from hypothesis import strategies as st from microcosm.build.telemetry_emitter import LocalTelemetryEmitter, TelemetryRun @@ -106,8 +111,13 @@ def __init__(self, rejected_run_ids: set[str] | None = None) -> None: self.requests: list[tuple[Run, str, Run | None, int]] = [] # (caller's run, Hugging Face token) for every refused exchange self.refused: list[tuple[Run, str]] = [] + # Runs once, while the next request is in flight. + self.during_next_request: Callable[[], None] | None = None def post(self, caller: Run, url: str, payload, token: str) -> tuple[int, dict]: + if self.during_next_request is not None: + during, self.during_next_request = self.during_next_request, None + during() path = urlsplit(url).path if path == TOKEN_EXCHANGE_PATH: if not token.startswith(MEMBER): @@ -181,8 +191,9 @@ def __init__(self, directory: Path, collector: FakeCollector) -> None: self.spool_path = directory / "events.sqlite3" self.collector = collector self.producers: list[Producer] = [] - # (caller's run, run made local-only, reason, caller's login then) - self.local_only: list[tuple[Run, Run, str, str | None]] = [] + # (caller's run, run made local-only, reason, caller's login then, + # number of collector requests by then) + self.local_only: list[tuple[Run, Run, str, str | None, int]] = [] # Requests a service sent about another producer's run while it lived. self.intrusions: list[tuple[Run, str, Run]] = [] @@ -200,7 +211,13 @@ def start(self, credential: str | None, run_id: str | None = None) -> Producer: def recorded(run_id: str, producer_id: str, reason: str) -> None: self.local_only.append( - (producer.run, (run_id, producer_id), reason, producer.credential) + ( + producer.run, + (run_id, producer_id), + reason, + producer.credential, + len(self.collector.requests), + ) ) make_local_only(run_id, producer_id, reason) @@ -276,7 +293,10 @@ def _assert_own_credential_only(host: Host) -> None: assert host.intrusions == [] rejected = host.collector.rejected_run_ids - for caller, run, reason, credential in host.local_only: + for caller, run, reason, credential, at in host.local_only: + # Local-only is final: nothing about the run reaches the collector after. + later = host.collector.requests[at:] + assert all(about != run for _, _, about, _ in later), (run, later) if caller != run: # Another producer's run: only the collector's answer about the # run's data, never one about this service's credential. @@ -319,6 +339,29 @@ def test_a_service_without_a_credential_leaves_a_live_build_run_pending(host): _assert_own_credential_only(host) +def test_a_run_made_local_only_after_it_was_listed_is_not_adopted(host): + """A pass lists the pending runs, then waits on the collector. Meanwhile + another build's service makes its own run local-only and exits, freeing + its lease. Found by the two-service test below.""" + + build_a = host.start(MEMBER + "alice") + build_b = host.start(None) + host.emit(build_a) + host.emit(build_b) + + def build_b_finishes() -> None: + host.flush(build_b) + host.kill(build_b) + + host.collector.during_next_request = build_b_finishes + host.flush(build_a) + + assert host.states()[build_b.run] == ("local_only", LOCAL_ONLY_MISSING_CREDENTIAL) + assert all(about != build_b.run for _, _, about, _ in host.collector.requests) + assert host.collector.accepted[build_a.run] == set(build_a.appended) + _assert_own_credential_only(host) + + @pytest.mark.parametrize("credential", CREDENTIALS) def test_a_live_build_run_gets_no_request_from_another_service(host, credential): build_a = host.start(MEMBER + "alice") @@ -611,12 +654,20 @@ def test_a_killed_service_releases_its_lease(tmp_path): process.stdout.close() +def _upload_state(spool_path: Path, run_id: str) -> str | None: + with closing(sqlite3.connect(spool_path)) as connection: + row = connection.execute( + "SELECT upload_state FROM telemetry_runs WHERE run_id = ?", (run_id,) + ).fetchone() + return row[0] if row else None + + def test_two_real_services_on_one_spool_leave_each_others_runs_alone( tmp_path, monkeypatch, real_local_telemetry ): """Build A has a credential, build B has none, and both services share one - spool. A's token exchange is held until B's worker has passed over A's - queued events several times; A's run must still arrive in full.""" + spool. A's token exchange is held until B's worker has made two passes + over A's queued run; A's run must still arrive in full.""" release = threading.Event() received: dict[str, list[dict[str, Any]]] = {} @@ -675,7 +726,13 @@ def log_message(self, format, *args): assert build_a.available and build_b.available build_a.stage("compile", message="Started.") build_b.stage("compile", message="Started.") - time.sleep(3.5) # B's worker passes over A's queued run every second + # B's first delivery pass, which meets A's queued run too, makes B's own + # run local-only. Let B make another pass, then let A deliver. + deadline = time.monotonic() + 60 + while _upload_state(spool_path, "build-b") != "local_only": + assert time.monotonic() < deadline, "build B's worker never ran" + time.sleep(0.1) + time.sleep(1.5) release.set() build_a.complete() build_b.complete() @@ -691,10 +748,6 @@ def log_message(self, format, *args): server_thread.join(timeout=5) server.server_close() - events = sorted(received["build-a"], key=lambda event: event["sequence"]) - assert [e["sequence"] for e in events] == list(range(1, len(events) + 1)) - assert events[-1]["message"] == BUILD_COMPLETED_MESSAGE - assert "build-b" not in received with closing(sqlite3.connect(spool_path)) as connection: states = dict( connection.execute( @@ -702,10 +755,18 @@ def log_message(self, format, *args): "FROM telemetry_runs" ) ) + # Before the fix, B's pass made A's run local-only (the e7 bug), and A, + # working from the list it read before, uploaded B's local-only run. assert states == { "build-a": "pending", "build-b": LOCAL_ONLY_MISSING_CREDENTIAL, } + assert "build-b" not in received, [ + (e["sequence"], e["message"]) for e in received["build-b"] + ] + events = sorted(received["build-a"], key=lambda event: event["sequence"]) + assert [e["sequence"] for e in events] == list(range(1, len(events) + 1)) + assert events[-1]["message"] == BUILD_COMPLETED_MESSAGE def _label_paths(host: Host) -> None: @@ -714,7 +775,7 @@ def _label_paths(host: Host) -> None: for caller, kind, about, status in host.collector.requests: if about is not None and about != caller: event(f"adopted orphan: {kind} {status}") - for caller, run, reason, _ in host.local_only: + for caller, run, reason, *_ in host.local_only: event(f"{'own' if caller == run else 'adopted'} run local-only: {reason}") @@ -731,6 +792,9 @@ 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)), + # 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()), ), max_size=30, ) @@ -756,6 +820,14 @@ def _start_build(host: Host, build: tuple[str | None, int]) -> Producer: operations=OPERATIONS, rejected=st.sets(st.integers(0, 7), max_size=2), ) +# Shrunk from a 1,500-example run against a delivery that trusted its listed +# runs: B makes its own run local-only and exits while A's exchange is in +# flight, and A then uploaded B's run. +@example( + builds=[(MEMBER + "alice", 0), (None, 0)], + operations=[("interleave", 0, 1, True)], + rejected=set(), +) def test_no_service_decides_another_producers_run_from_its_own_credential( tmp_path_factory, builds, operations, rejected ): @@ -782,8 +854,21 @@ def test_no_service_decides_another_producers_run_from_its_own_credential( host.flush(producer) elif kind == "kill": host.kill(producer) - else: + elif kind == "login": producer.credential = operation[2] + else: + _, _, other_index, exits = operation + other = live[other_index % len(live)] + if other is not producer: + + def other_flushes(other=other, exits=exits) -> None: + host.flush(other) + if exits: + host.kill(other) + + host.collector.during_next_request = other_flushes + host.flush(producer) + host.collector.during_next_request = None _assert_own_credential_only(host) _label_paths(host) From 1c9bd671b0711b6c7b920bc9755361f4785a948b Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Fri, 9 Oct 2026 21:03:29 -0400 Subject: [PATCH 3/4] Fix four defects from review of the foreign-run delivery Review round 1 (GPT-6.1 Sol, at 662fd4acb) reproduced four defects. Each has a test that fails on 662fd4acb and passes now. 1. A service that could not lock its own lease still served. The first lock error ended the attempt, and the attempt had already created the lease file. Other services read a lease file nobody holds as an exited producer's, so they adopted the live run and registered it under their own login. Now the attempt is repeated until its deadline, and a service that still has no lease does not start: the build goes on without telemetry and says why. A platform with no advisory locks is unchanged. 2. A run passed over under a login the collector refused was never offered again after the login changed. With the service's own run local-only and every other pending run passed over, nothing read the login, so the pass-over check kept comparing against the old one. Each pass with passed-over runs now notes the credential its requests would use. A session the collector issued stays in use until it expires, as for the own run. 3. The delivery's per-run memory was keyed by "run_id:producer_id". Both ids may contain colons, so two runs could share a key: a refusal for one passed the other over, and registering one counted for the other. The key is now the id pair. 4. leases._lock left its descriptor open, and the lock held, when the inode check itself raised. Tests: the property test's last two services now start under generated logins, make a pass, and only then become Alice and Bob with their sessions expired, so recovery by a service that was refused first is checked. It also generates run ids the collector answers with 409. A local-only decision is checked against the deciding service's latest collector answer, not any earlier one. The fake clock the session tests need replaces time.monotonic in the delivery during each flush. Co-Authored-By: Claude Opus 5.5 --- .../telemetry-foreign-run-local-only.fixed.md | 2 +- .../telemetry_emitter_service/collector.py | 114 +++-- .../telemetry_emitter_service/constants.py | 5 + .../build/telemetry_emitter_service/leases.py | 27 +- .../build/telemetry_emitter_service/main.py | 20 +- .../shared/test_telemetry_foreign_runs.py | 441 ++++++++++++++++-- 6 files changed, 527 insertions(+), 82 deletions(-) diff --git a/changelog.d/telemetry-foreign-run-local-only.fixed.md b/changelog.d/telemetry-foreign-run-local-only.fixed.md index ad2b438018..87adb7e43c 100644 --- a/changelog.d/telemetry-foreign-run-local-only.fixed.md +++ b/changelog.d/telemetry-foreign-run-local-only.fixed.md @@ -1 +1 @@ -A build's telemetry service no longer disables hosted telemetry for other builds on the same machine. Every service shares one local spool. Before, a service with no Hugging Face login, or with a login the collector refused, permanently marked every other build's queued run local-only. Now a service touches another build's run only after that build's service has exited, and only one service does so at a time. Each service holds an advisory lock file next to the spool. A run with no lock file, from an older checkout, waits until it has been idle for 15 minutes. A missing, refused or non-owning login decides only the service's own run; another build's run stays pending for a service whose login the collector accepts for it. A refused login is sent to the collector once, not on every pass, and the shutdown drain waits only for the build's own run. +A build's telemetry service no longer disables hosted telemetry for other builds on the same machine. Every service shares one local spool. Before, a service with no Hugging Face login, or with a login the collector refused, permanently marked every other build's queued run local-only. Now a service touches another build's run only after that build's service has exited, and only one service does so at a time. Each service holds an advisory lock file next to the spool. A service that cannot take its lock does not start, and that build continues without telemetry. A run with no lock file, from an older checkout, waits until it has been idle for 15 minutes. A missing, refused or non-owning login decides only the service's own run; another build's run stays pending for a service whose login the collector accepts for it, and a service that was refused for it offers it again once its own login changes. A refused login is sent to the collector once, not on every pass, and the shutdown drain waits only for the build's own run. 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 12f1cf9bd0..979d26a078 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 @@ -48,7 +48,11 @@ TOKEN_EXCHANGE_PATH, TOKEN_REFRESH_MARGIN_SECONDS, ) -from microcosm.build.telemetry_emitter_service.leases import ProducerLeases +from microcosm.build.telemetry_emitter_service.leases import ( + OwnLeaseUnavailableError, + ProducerLease, + ProducerLeases, +) from microcosm.build.telemetry_emitter_service.spool import EventSpool @@ -144,11 +148,10 @@ def _permanent_client_error(status: int) -> bool: return 400 <= int(status) < 500 and status not in _RETRYABLE_CLIENT_ERRORS -def _registration_key(registration: Mapping[str, Any]) -> str: - return f"{registration['run_id']}:{registration['producer_id']}" - - 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.""" + return str(registration["run_id"]), str(registration["producer_id"]) @@ -191,14 +194,11 @@ def __init__( ) # Taken before the service is ready, so before its run has any event: # no other service can find the run pending without a live lease. - self._own_lease = self._leases.hold( - *self._own_run, - timeout_seconds=OWN_LEASE_TIMEOUT_SECONDS, - ) + self._own_lease = self._hold_own_lease() self._session_token: tuple[str, float] | None = None - self._registered: set[str] = set() + self._registered: set[tuple[str, str]] = set() self._warned_no_token = False - self._warned_denied: set[str] = set() + self._warned_denied: set[tuple[str, str]] = set() self._next_attempt_at = 0.0 self._retry_seconds = INITIAL_RETRY_SECONDS self._next_lease_sweep_at = 0.0 @@ -208,7 +208,26 @@ def __init__( self._rejected_credentials: set[str] = set() # Other producers' runs this credential cannot deliver, by the # credential that was refused; tried again once the credential changes. - self._passed_over: dict[str, str | None] = {} + self._passed_over: dict[tuple[str, str], str | None] = {} + + def _hold_own_lease(self) -> ProducerLease | None: + """Take this service's lease, or refuse to deliver without it. + + Other services read a lease file nobody holds as an exited producer's, + and a failed attempt can leave one behind. A service that served + without its lease could so have its live run adopted, and registered + under another login. Only where there are no advisory locks at all does + a service run without one; other services then judge its run by recent + activity. + """ + + lease = self._leases.hold( + *self._own_run, + timeout_seconds=OWN_LEASE_TIMEOUT_SECONDS, + ) + if lease is None and self._leases.supported: + raise OwnLeaseUnavailableError(self._leases.path(*self._own_run)) + return lease def close(self) -> None: """Release this service's lease; its run, if still pending, is adoptable.""" @@ -223,12 +242,14 @@ def flush_once(self) -> bool: return False made_progress = False pending = self.spool.pending_runs() - pending_keys = {_registration_key(registration) for registration in pending} + pending_runs = {_run(registration) for registration in pending} self._passed_over = { - key: credential - for key, credential in self._passed_over.items() - if key in pending_keys + run: credential + for run, credential in self._passed_over.items() + if run in pending_runs } + if self._passed_over: + self._note_current_credential() self._sweep_leases_if_due(pending) for registration in pending: claim = self._claim(registration) @@ -261,8 +282,7 @@ def _claim( run = _run(registration) if run == self._own_run: return contextlib.nullcontext() - key = _registration_key(registration) - if key in self._passed_over and self._passed_over[key] == self._credential: + if run in self._passed_over and self._passed_over[run] == self._credential: return None if not self._leases.supported: return contextlib.nullcontext() if self._idle(run) else None @@ -277,6 +297,29 @@ def _claim( # A lease that cannot be tested may be live: leave the run be. return None + def _note_current_credential(self) -> None: + """Record the credential this pass's requests would be sent under. + + A run passed over with one credential is offered again once the login + changes. Delivering a run is what reads the login, and a passed-over run + is not delivered, so with nothing else pending the change would go + unseen. A session the collector issued stays in use until it expires, + as for the own run, so only then does a new login count. + """ + + if self._cached_session_token() is not None: + return + hf_token = _huggingface_token() + self._credential = _credential_fingerprint(hf_token) if hf_token else None + + def _cached_session_token(self) -> str | None: + if ( + self._session_token is not None + and self._session_token[1] > time.monotonic() + TOKEN_REFRESH_MARGIN_SECONDS + ): + return self._session_token[0] + return None + def _idle(self, run: tuple[str, str]) -> bool: updated_at = self.spool.last_updated(*run) if updated_at is None: @@ -296,15 +339,14 @@ def _sweep_leases_if_due(self, pending: list[dict[str, Any]]) -> None: def _deliver(self, registration: Mapping[str, Any]) -> bool: """Send one batch of a run's events; return whether any was delivered.""" - run_id = str(registration["run_id"]) - registration_key = _registration_key(registration) + run_id, producer_id = _run(registration) token = self._collector_token(registration) if token is None: return False - if registration_key not in self._registered: + if (run_id, producer_id) not in self._registered: if not self._register(registration, token): return False - events = self.spool.batch(run_id, str(registration["producer_id"])) + events = self.spool.batch(run_id, producer_id) if not events: return False try: @@ -342,11 +384,9 @@ def _deliver(self, registration: Mapping[str, Any]) -> bool: return False def _collector_token(self, registration: Mapping[str, Any]) -> str | None: - if ( - self._session_token is not None - and self._session_token[1] > time.monotonic() + TOKEN_REFRESH_MARGIN_SECONDS - ): - return self._session_token[0] + session_token = self._cached_session_token() + if session_token is not None: + return session_token hf_token = _huggingface_token() if not hf_token: self._credential = None @@ -399,7 +439,6 @@ def _collector_token(self, registration: Mapping[str, Any]) -> str | None: return None def _register(self, registration: Mapping[str, Any], token: str) -> bool: - registration_key = _registration_key(registration) try: status, _ = _http_post( self.collector_url + RUN_REGISTRATION_PATH, @@ -410,7 +449,7 @@ def _register(self, registration: Mapping[str, Any], token: str) -> bool: self._defer_retry() return False if status == HTTPStatus.CREATED: - self._registered.add(registration_key) + self._registered.add(_run(registration)) self._retry_seconds = INITIAL_RETRY_SECONDS return True if status == HTTPStatus.UNAUTHORIZED: @@ -451,25 +490,24 @@ def _make_local_only( and is not offered again with this credential. """ - run_id, producer_id = _run(registration) - registration_key = _registration_key(registration) - own = (run_id, producer_id) == self._own_run + run = _run(registration) + own = run == self._own_run if caller_dependent and not own: if reason != LOCAL_ONLY_MISSING_CREDENTIAL: - self._passed_over[registration_key] = self._credential + self._passed_over[run] = self._credential return - self.spool.make_local_only(run_id, producer_id, reason) + 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: - self._warn_denied(registration_key, message) + 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, registration_key: str, message: str) -> None: - if registration_key in self._warned_denied: + def _warn_denied(self, run: tuple[str, str], message: str) -> None: + if run in self._warned_denied: return print(message, file=sys.stderr, flush=True) - self._warned_denied.add(registration_key) + self._warned_denied.add(run) 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 362ef8ac23..e4331b9039 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 @@ -42,6 +42,11 @@ "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 " + "could upload the run under its own login. The dataset build will continue." +) COLLECTOR_URL_HTTPS_ERROR: Final = "collector URL must be an HTTPS origin" COLLECTOR_URL_ORIGIN_ERROR: Final = ( "collector URL must be an origin without credentials or path data" diff --git a/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/leases.py b/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/leases.py index 60f76aeffe..6fe562b7e7 100644 --- a/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/leases.py +++ b/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/leases.py @@ -36,6 +36,10 @@ ) +class OwnLeaseUnavailableError(RuntimeError): + """A service could not take its own producer's lease.""" + + class ProducerLease: """An exclusive lock on one producer's lease file, held until released.""" @@ -135,20 +139,25 @@ def hold( clock: Callable[[], float] = time.monotonic, sleep: Callable[[float], None] = time.sleep, ) -> ProducerLease | None: - """Take this service's own lease, waiting out brief holders. + """Take this service's own lease, waiting out brief holders and errors. Only a sweeper or an adopter holds another producer's lease, and only - for a moment, so a short wait suffices. Returns ``None`` if the lease is - still taken at the deadline or cannot be taken at all; the service then - runs without one, and other services judge its run by recent activity. + for a moment, and a lock the system refuses once may be granted on the + next attempt, so a short wait suffices. Returns ``None`` if the lease is + still not held at the deadline, or on a platform without advisory + locks. A failed attempt can leave a lease file nobody holds, which + reads as an exited producer's, so the caller must not serve its run + without the lease. """ + if fcntl is None: + return None deadline = clock() + timeout_seconds while True: try: lease = self.try_acquire(run_id, producer_id, create=True) except OSError: - return None + lease = None if lease is not None or clock() >= deadline: return lease sleep(LEASE_RETRY_SECONDS) @@ -192,7 +201,13 @@ def _lock(path: Path, *, create: bool) -> ProducerLease | None: except BaseException: os.close(descriptor) raise - if _names_open_file(path, descriptor): + try: + current = _names_open_file(path, descriptor) + except BaseException: + # Never leave a lock held on a descriptor nobody will close. + os.close(descriptor) + raise + if current: return ProducerLease(path, descriptor) # Another process removed this file between our open and our lock. os.close(descriptor) diff --git a/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/main.py b/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/main.py index 9216f09569..c8a5fffd33 100644 --- a/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/main.py +++ b/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/main.py @@ -4,12 +4,15 @@ import argparse import json +import sys from pathlib import Path from microcosm.build.telemetry_emitter_service.collector import CollectorDelivery from microcosm.build.telemetry_emitter_service.constants import ( DEFAULT_HEARTBEAT_SECONDS, + OWN_LEASE_UNAVAILABLE_MESSAGE, ) +from microcosm.build.telemetry_emitter_service.leases import OwnLeaseUnavailableError from microcosm.build.telemetry_emitter_service.resources import ProcessTreeSampler from microcosm.build.telemetry_emitter_service.runtime import EmitterService from microcosm.build.telemetry_emitter_service.spool import EventSpool @@ -39,12 +42,17 @@ def main(argv: list[str] | None = None) -> int: registration = json.loads(args.registration_json) spool = EventSpool(args.spool) # The delivery takes this producer's lease, which must happen before the - # service is ready: from then on the run can have events. - delivery = CollectorDelivery( - spool, - registration, - development_collector_url=args.development_collector_url, - ) + # service is ready: from then on the run can have events. Without the + # lease the service does not start, and the build goes on without it. + try: + delivery = CollectorDelivery( + spool, + registration, + development_collector_url=args.development_collector_url, + ) + except OwnLeaseUnavailableError: + print(OWN_LEASE_UNAVAILABLE_MESSAGE, file=sys.stderr, flush=True) + return 1 service = EmitterService( socket_path=args.socket, registration=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 0f44953245..a587d4ab05 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 @@ -17,18 +17,22 @@ run goes local-only only after a collector answer about the run itself. - While a producer's service is alive, no other service sends any request for that producer's run. -- A run goes local-only only right after a collector answer (or, for a missing - credential, the absence of one) that justifies it, and a refused credential - is exchanged once. +- A run goes local-only only when the deciding service's latest collector + answer is the refusal that justifies it. Two cases have no fresh answer: a + missing credential, and a credential the collector already refused to that + service, which is exchanged once and not again. - Local-only is final: no request about a run reaches the collector after it 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. + service whose credential owns it, including a service that was refused for + the run under an earlier login. """ from __future__ import annotations +import errno import json +import os import signal import sqlite3 import subprocess @@ -59,17 +63,24 @@ ) from microcosm.build.telemetry_emitter_service import collector as collector_module 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 ( + DEFAULT_TOKEN_LIFETIME_SECONDS, 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, + OWN_LEASE_UNAVAILABLE_MESSAGE, RUN_REGISTRATION_PATH, TOKEN_EXCHANGE_PATH, ) -from microcosm.build.telemetry_emitter_service.leases import ProducerLeases +from microcosm.build.telemetry_emitter_service.leases import ( + OwnLeaseUnavailableError, + ProducerLeases, +) from microcosm.build.telemetry_protocol import BUILD_COMPLETED_MESSAGE MEMBER = "hf-member:" @@ -98,22 +109,45 @@ def _registration(run_id: str, producer_id: str) -> dict[str, Any]: ).as_registration() +def _run_of(registration: dict[str, Any]) -> Run: + return registration["run_id"], registration["producer_id"] + + class FakeCollector: """The hosted collector's answers, as its authorization model gives them.""" - def __init__(self, rejected_run_ids: set[str] | None = None) -> None: + def __init__( + self, + rejected_run_ids: set[str] | None = None, + conflicted_run_ids: set[str] | None = None, + ) -> None: self.rejected_run_ids = set(rejected_run_ids or ()) self.owners: dict[str, str] = {} - self.metadata: dict[str, dict[str, Any]] = {} + # A conflicted run id was registered before with other metadata, so + # whoever owns it gets 409 for this registration. + self.metadata: dict[str, dict[str, Any]] = { + run_id: {"pipeline": "something-else"} + for run_id in conflicted_run_ids or () + } self.producers: set[Run] = set() self.accepted: dict[Run, set[str]] = {} # (caller's run, kind, run asked about or None, status) self.requests: list[tuple[Run, str, Run | None, int]] = [] - # (caller's run, Hugging Face token) for every refused exchange + # (caller's run, Hugging Face token) for every refused exchange, and + # where in ``requests`` the first refusal of each falls self.refused: list[tuple[Run, str]] = [] + self.refused_at: dict[tuple[Run, str], int] = {} # Runs once, while the next request is in flight. self.during_next_request: Callable[[], None] | None = None + @property + def conflicted_run_ids(self) -> set[str]: + return { + run_id + for run_id, metadata in self.metadata.items() + if metadata.get("pipeline") == "something-else" + } + def post(self, caller: Run, url: str, payload, token: str) -> tuple[int, dict]: if self.during_next_request is not None: during, self.during_next_request = self.during_next_request, None @@ -122,6 +156,7 @@ def post(self, caller: Run, url: str, payload, token: str) -> tuple[int, dict]: if path == TOKEN_EXCHANGE_PATH: if not token.startswith(MEMBER): self.refused.append((caller, token)) + self.refused_at.setdefault((caller, token), len(self.requests)) if token.startswith(MEMBER): user = token.removeprefix(MEMBER) return self._answer( @@ -160,6 +195,14 @@ def _answer(self, caller, kind, run, status, body): self.requests.append((caller, kind, run, status)) return status, body + 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.""" + + for who, kind, about, status in reversed(self.requests[:before]): + if who == caller: + return kind, about, status + return None + def asked(self, caller: Run, kind: str, run: Run | None = None) -> list[int]: return [ status @@ -181,7 +224,7 @@ class Producer: @property def run(self) -> Run: - return self.registration["run_id"], self.registration["producer_id"] + return _run_of(self.registration) class Host: @@ -196,10 +239,20 @@ def __init__(self, directory: Path, collector: FakeCollector) -> None: self.local_only: list[tuple[Run, Run, str, str | None, int]] = [] # Requests a service sent about another producer's run while it lived. self.intrusions: list[tuple[Run, str, Run]] = [] - - def start(self, credential: str | None, run_id: str | None = None) -> Producer: + # Each flush comes after any retry delay a service set itself, as the + # service's one-second worker ticks would reach it. + self.clock = 0.0 + + def start( + self, + credential: str | None, + run_id: str | None = None, + producer_id: str | None = None, + ) -> Producer: index = len(self.producers) - registration = _registration(run_id or f"run-{index}", f"producer-{index}") + registration = _registration( + run_id or f"run-{index}", producer_id or f"producer-{index}" + ) spool = EventSpool(self.spool_path) # As in the service's main(): the lease is taken before registration. delivery = CollectorDelivery( @@ -242,11 +295,15 @@ def post(url, payload, token, *, timeout=5.0): self.intrusions.append((producer.run, url, about)) return status, body + self.clock += MAX_RETRY_SECONDS + 1 with ( mock.patch.object( collector_module, "_huggingface_token", lambda: producer.credential ), mock.patch.object(collector_module, "_http_post", post), + mock.patch.object( + collector_module, "time", SimpleNamespace(monotonic=lambda: self.clock) + ), ): return producer.delivery.flush_once() @@ -293,28 +350,44 @@ def _assert_own_credential_only(host: Host) -> None: 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. later = host.collector.requests[at:] assert all(about != run for _, _, about, _ in later), (run, later) + # What this service last heard from the collector before deciding. + latest = host.collector.latest(caller, at) if caller != run: # Another producer's run: only the collector's answer about the - # run's data, never one about this service's credential. - assert reason == LOCAL_ONLY_REJECTED_EVENTS, (caller, run, reason) - assert run[0] in rejected - assert 422 in host.collector.asked(caller, "events", run) + # run itself, never one about this service's credential. + if reason == LOCAL_ONLY_REJECTED_REGISTRATION: + assert run[0] in conflicted, (caller, run) + 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) continue if reason == LOCAL_ONLY_MISSING_CREDENTIAL: assert credential is None elif reason == LOCAL_ONLY_REJECTED_CREDENTIAL: - assert set(host.collector.asked(caller, "exchange")) & {401, 403} + # The collector refused this very login to this service, just now + # or earlier: a refused login is not exchanged a second time. + first_refused = host.collector.refused_at.get((caller, credential), at) + assert first_refused < at, (caller, credential) elif reason == LOCAL_ONLY_REJECTED_REGISTRATION: - assert set(host.collector.asked(caller, "register", run)) & {403, 409} + assert latest in (("register", run, 403), ("register", run, 409)), ( + run, + latest, + ) elif reason == LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION: - assert 403 in host.collector.asked(caller, "events", run) + assert latest == ("events", run, 403), (run, latest) else: assert reason == LOCAL_ONLY_REJECTED_EVENTS - assert set(host.collector.asked(caller, "events", run)) & {404, 422} + assert latest in (("events", run, 404), ("events", run, 422)), ( + 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 @@ -435,6 +508,260 @@ def test_an_orphan_owned_by_another_user_is_passed_over_not_made_local_only(host _assert_own_credential_only(host) +@pytest.mark.parametrize("refused", [OUTSIDER, EXPIRED]) +def test_a_passed_over_orphan_is_offered_again_once_the_login_changes(host, refused): + """Once a service's own run is local-only and every other pending run is + passed over, no delivery reads the login any more. The pass itself must + notice the new login, or the orphan waits for another service.""" + + build_a = host.start(MEMBER + "alice") + host.emit(build_a) + host.flush(build_a) + host.emit(build_a, 2) + host.kill(build_a) + build_b = host.start(refused) + host.emit(build_b) + for _ in range(2): + host.flush(build_b) + assert host.states()[build_b.run] == ( + "local_only", + LOCAL_ONLY_REJECTED_CREDENTIAL, + ) + assert host.queued(build_a.run) == 2 + + build_b.credential = MEMBER + "alice" + host.flush(build_b) + + assert host.collector.accepted[build_a.run] == set(build_a.appended) + assert host.queued(build_a.run) == 0 + # The refused login was exchanged once, and so was the new one. + assert host.collector.asked(build_b.run, "exchange") == [ + 403 if refused == OUTSIDER else 401, + 200, + ] + _assert_own_credential_only(host) + + +def test_a_passed_over_orphan_waits_out_the_session_of_the_login_that_was_refused( + host, +): + """A session the collector issued stays in use until it expires, so a new + login takes effect then, as it does for the service's own run.""" + + build_a = host.start(MEMBER + "alice") + host.emit(build_a) + host.flush(build_a) + host.emit(build_a, 2) + host.kill(build_a) + build_b = host.start(MEMBER + "bob") + host.emit(build_b) + host.flush(build_b) + assert host.collector.asked(build_b.run, "register", build_a.run) == [403] + + build_b.credential = MEMBER + "alice" + host.flush(build_b) + # Bob's session is still what this service sends, and it was refused. + assert host.collector.asked(build_b.run, "register", build_a.run) == [403] + assert host.queued(build_a.run) == 2 + + host.clock += DEFAULT_TOKEN_LIFETIME_SECONDS + host.flush(build_b) + + assert host.collector.asked(build_b.run, "register", build_a.run) == [403, 201] + assert host.collector.accepted[build_a.run] == set(build_a.appended) + _assert_own_credential_only(host) + + +def test_a_refusal_for_one_run_does_not_pass_over_a_run_with_similar_ids(host): + """Run and producer ids may contain colons, so two runs can share the text + ``run_id:producer_id``. From review: Bob was refused Alice's ``a:b``/``c`` + and then never offered his own ``a``/``b:c``.""" + + build_a = host.start(MEMBER + "alice", run_id="a:b", producer_id="c") + host.emit(build_a) + host.flush(build_a) + host.emit(build_a) + host.kill(build_a) + build_b = host.start(MEMBER + "bob", run_id="a", producer_id="b:c") + host.emit(build_b) + host.flush(build_b) + host.emit(build_b) + host.kill(build_b) + adopter = host.start(MEMBER + "bob") + + for _ in range(2): + host.flush(adopter) + + assert host.collector.asked(adopter.run, "register", build_a.run) == [403] + assert host.collector.accepted[build_b.run] == set(build_b.appended) + assert host.queued(build_b.run) == 0 + assert host.queued(build_a.run) == 1 + _assert_own_credential_only(host) + + +def test_registering_one_run_does_not_count_for_a_run_with_similar_ids(host): + """Both orphans are Alice's and neither was registered. The adopter must + register each: the second is not registered just because the first's ids + join to the same text.""" + + orphans = [ + host.start(MEMBER + "alice", run_id="a:b", producer_id="c"), + host.start(MEMBER + "alice", run_id="a", producer_id="b:c"), + ] + for orphan in orphans: + host.emit(orphan) + host.kill(orphan) + adopter = host.start(MEMBER + "alice") + + for _ in range(2): + host.flush(adopter) + + for orphan in orphans: + assert host.collector.asked(adopter.run, "register", orphan.run) == [201] + assert host.collector.accepted[orphan.run] == set(orphan.appended) + assert host.states()[orphan.run] == ("pending", None) + _assert_own_credential_only(host) + + +def _refuse_locks(patch, refusals: int | None) -> list[int]: + """Make the next ``refusals`` lock attempts fail (all of them, if ``None``) + as a system that has run out of locks would.""" + + real = leases_module.fcntl + attempts: list[int] = [] + + def flock(descriptor: int, operation: int) -> None: + attempts.append(descriptor) + if refusals is None or len(attempts) <= refusals: + raise OSError(errno.ENOLCK, "No locks available") + real.flock(descriptor, operation) + + patch.setattr( + leases_module, + "fcntl", + SimpleNamespace(flock=flock, LOCK_EX=real.LOCK_EX, LOCK_NB=real.LOCK_NB), + ) + return attempts + + +def test_a_lock_refused_once_at_startup_does_not_expose_the_run(host, monkeypatch): + """From review: the system refused build A's lock once, so A ran without + its lease, and the attempt left a lease file nobody held. Build B took + that for an orphan's and registered A's live run under its own login.""" + + with monkeypatch.context() as patch: + attempts = _refuse_locks(patch, 1) + build_a = host.start(MEMBER + "alice") + build_b = host.start(MEMBER + "bob") + host.emit(build_a) + host.emit(build_b) + + for _ in range(2): + host.flush(build_b) + + assert all(about != build_a.run for _, _, about, _ in host.collector.requests) + host.flush(build_a) + assert host.collector.accepted[build_a.run] == set(build_a.appended) + assert host.states()[build_a.run] == ("pending", None) + _assert_own_credential_only(host) + # A took its lease on the attempt after the refusal. + assert len(attempts) == 2 + + +@pytest.mark.parametrize("obstacle", ["locks_refused", "lease_held"]) +def test_a_service_that_cannot_take_its_lease_does_not_start( + tmp_path, monkeypatch, capsys, obstacle +): + """A lease file nobody holds reads as an exited producer's, and a failed + attempt can leave one. So a service that cannot take its lease serves + nothing: its run never has an event another service could adopt.""" + + registration = _registration("run-a", "producer-a") + spool_path = tmp_path / "events.sqlite3" + arguments = [ + *("--socket", str(tmp_path / "service.sock")), + *("--spool", str(spool_path)), + *("--development-collector-url", DEVELOPMENT_COLLECTOR), + *("--registration-json", json.dumps(registration)), + *("--parent-pid", str(os.getpid())), + ] + served: list[EmitterService] = [] + monkeypatch.setattr(EmitterService, "run", lambda self: served.append(self)) + spools: list[EventSpool] = [] + + def opened(path: Path) -> EventSpool: + spools.append(EventSpool(path)) + return spools[-1] + + monkeypatch.setattr(main_module, "EventSpool", opened) + monkeypatch.setattr(collector_module, "OWN_LEASE_TIMEOUT_SECONDS", 0.05) + try: + with monkeypatch.context() as patch: + if obstacle == "locks_refused": + _refuse_locks(patch, None) + holder = None + else: + # An adopter is still delivering an earlier run with these ids. + holder = ProducerLeases.beside(spool_path).try_acquire( + *_run_of(registration), create=True + ) + assert holder is not None + with pytest.raises(OwnLeaseUnavailableError): + CollectorDelivery( + opened(spool_path), + registration, + development_collector_url=DEVELOPMENT_COLLECTOR, + ) + assert main_module.main(arguments) == 1 + assert served == [] + assert capsys.readouterr().err.strip() == OWN_LEASE_UNAVAILABLE_MESSAGE + if holder is not None: + holder.release() + + # With the obstacle gone, the same service starts. + assert main_module.main(arguments) == 0 + assert [service.registration for service in served] == [registration] + assert capsys.readouterr().err == "" + finally: + for spool in spools: + spool._engine.dispose() + + +def test_an_own_lease_waits_out_refused_locks_until_its_deadline(tmp_path, monkeypatch): + leases = ProducerLeases(tmp_path / "leases") + now = [0.0] + + def sleep(seconds: float) -> None: + now[0] += seconds + + def hold(): + return leases.hold( + "run-a", + "producer-a", + timeout_seconds=1, + clock=lambda: now[0], + sleep=sleep, + ) + + with monkeypatch.context() as patch: + attempts = _refuse_locks(patch, 3) + lease = hold() + assert lease is not None + assert len(attempts) == 4 + lease.release() + + with monkeypatch.context() as patch: + attempts = _refuse_locks(patch, None) + assert hold() is None + # It kept asking until the deadline rather than giving up at once. + assert len(attempts) > 50 + + # The refused attempts left nothing locked. + lease = leases.try_acquire("run-a", "producer-a", create=False) + assert lease is not None + lease.release() + + @pytest.mark.parametrize("answer", ["unacceptable_events", "registration_conflict"]) def test_an_orphan_the_collector_refuses_goes_local_only_without_a_warning_here( host, capsys, answer @@ -619,6 +946,24 @@ def flock(descriptor: int, operation: int) -> None: leases.try_acquire("run-a", "producer-a", create=False) +def test_a_lease_that_cannot_be_checked_is_not_left_locked(tmp_path, monkeypatch): + leases = ProducerLeases(tmp_path / "leases") + leases.try_acquire("run-a", "producer-a", create=True).release() + + def unreadable(path, descriptor): + raise PermissionError("cannot stat the lease file") + + with monkeypatch.context() as patch: + patch.setattr(leases_module, "_names_open_file", unreadable) + with pytest.raises(PermissionError): + leases.try_acquire("run-a", "producer-a", create=False) + + # The failed attempt closed its descriptor, so the lock went with it. + lease = leases.try_acquire("run-a", "producer-a", create=False) + assert lease is not None + lease.release() + + def test_a_killed_service_releases_its_lease(tmp_path): leases = ProducerLeases(tmp_path / "leases") holder = textwrap.dedent( @@ -666,8 +1011,9 @@ def test_two_real_services_on_one_spool_leave_each_others_runs_alone( tmp_path, monkeypatch, real_local_telemetry ): """Build A has a credential, build B has none, and both services share one - spool. A's token exchange is held until B's worker has made two passes - over A's queued run; A's run must still arrive in full.""" + spool. A's token exchange is held until B's worker has made a pass over + A's queued run, and for 1.5 s more (the worker passes every second); A's + run must still arrive in full.""" release = threading.Event() received: dict[str, list[dict[str, Any]]] = {} @@ -727,7 +1073,7 @@ def log_message(self, format, *args): build_a.stage("compile", message="Started.") build_b.stage("compile", message="Started.") # B's first delivery pass, which meets A's queued run too, makes B's own - # run local-only. Let B make another pass, then let A deliver. + # run local-only. Leave time for another pass, then let A deliver. deadline = time.monotonic() + 60 while _upload_state(spool_path, "build-b") != "local_only": assert time.monotonic() < deadline, "build B's worker never ran" @@ -819,6 +1165,9 @@ def _start_build(host: Host, build: tuple[str | None, int]) -> Producer: builds=st.lists(BUILD, min_size=2, max_size=4), operations=OPERATIONS, rejected=st.sets(st.integers(0, 7), max_size=2), + conflicted=st.sets(st.integers(0, 7), max_size=2), + # 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)), ) # Shrunk from a 1,500-example run against a delivery that trusted its listed # runs: B makes its own run local-only and exits while A's exchange is in @@ -827,13 +1176,27 @@ def _start_build(host: Host, build: tuple[str | None, int]) -> Producer: builds=[(MEMBER + "alice", 0), (None, 0)], operations=[("interleave", 0, 1, True)], rejected=set(), + conflicted=set(), + rescue_logins=(MEMBER + "alice", MEMBER + "bob"), +) +# From review: a service refused for every pending run, whose own run is not +# pending either, never read its login again, so a new login went unseen. +@example( + builds=[(MEMBER + "alice", 0), (MEMBER + "alice", 0)], + operations=[], + rejected=set(), + conflicted=set(), + rescue_logins=(OUTSIDER, EXPIRED), ) def test_no_service_decides_another_producers_run_from_its_own_credential( - tmp_path_factory, builds, operations, rejected + tmp_path_factory, builds, operations, rejected, conflicted, rescue_logins ): host = Host( tmp_path_factory.mktemp("foreign"), - FakeCollector({f"run-{index}" for index in rejected}), + FakeCollector( + {f"run-{index}" for index in rejected}, + {f"run-{index}" for index in conflicted}, + ), ) try: # Builds already running, each with an event queued. @@ -870,26 +1233,42 @@ def other_flushes(other=other, exits=exits) -> None: host.flush(producer) host.collector.during_next_request = None _assert_own_credential_only(host) - _label_paths(host) - # Every service goes. A run still pending is an orphan, and a service - # whose login owns it delivers it in full. + # Every service goes. A run still pending is an orphan. Two more + # services start under any login and make a pass, which may refuse + # them. Then their logins become Alice's and Bob's and any session + # they held runs out: between them they deliver each orphan in full. for producer in host.producers: if producer.alive: host.kill(producer) - rescuers = [host.start(MEMBER + user) for user in ("alice", "bob")] before = host.states() + rescuers = [host.start(login) for login in rescue_logins] + for rescuer in rescuers: + host.flush(rescuer) + _assert_own_credential_only(host) + for rescuer, user in zip(rescuers, ("alice", "bob"), strict=True): + rescuer.credential = MEMBER + user + host.clock += DEFAULT_TOKEN_LIFETIME_SECONDS for _ in range(3): for rescuer in rescuers: host.flush(rescuer) + _label_paths(host) _assert_own_credential_only(host) after = host.states() for producer in host.producers: if producer in rescuers or before[producer.run][0] != "pending": continue - if producer.run[0] in host.collector.rejected_run_ids and ( - producer.appended + run_id = producer.run[0] + if host.queued(producer.run) and ( + run_id in host.collector.conflicted_run_ids ): + # Its owner, whoever that is, gets 409: a refusal about the run. + assert after[producer.run] == ( + "local_only", + LOCAL_ONLY_REJECTED_REGISTRATION, + ), producer.run + continue + if run_id in host.collector.rejected_run_ids and producer.appended: if host.queued(producer.run): assert after[producer.run] == ( "local_only", From 615d84f899b6d48144a6819e3103e85e334130d6 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Fri, 9 Oct 2026 21:39:03 -0400 Subject: [PATCH 4/4] Test an adopted run's events 403 and pin the lease file layout Two tests the microcosm-emitter port has and this branch lacked: - an adopted run whose events the collector refuses with 403 (another user owns it) is passed over, not made local-only, and its owner delivers it; - the lease directory, file name, digest and modes are pinned to one id pair, since services from different versions of either package must agree on them to see each other's live runs. Co-Authored-By: Claude Opus 5.5 --- .../shared/test_telemetry_foreign_runs.py | 46 +++++++++++++++++++ 1 file changed, 46 insertions(+) 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 a587d4ab05..7ba5224d84 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 @@ -508,6 +508,32 @@ def test_an_orphan_owned_by_another_user_is_passed_over_not_made_local_only(host _assert_own_credential_only(host) +def test_an_orphan_whose_events_another_user_may_not_send_is_passed_over(host): + """An events 403 is about the caller too: the run is another user's.""" + + build_a = host.start(MEMBER + "alice") + host.emit(build_a) + host.flush(build_a) + host.emit(build_a, 2) + host.kill(build_a) + build_b = host.start(MEMBER + "bob") + # Bob's service remembers registering the run, as it would if its login + # changed from Alice's to Bob's afterwards, so it goes straight to the + # events and is refused there. + build_b.delivery._registered.add(build_a.run) + + for _ in range(3): + host.flush(build_b) + + assert host.collector.asked(build_b.run, "events", build_a.run) == [403] + assert host.states()[build_a.run] == ("pending", None) + assert host.queued(build_a.run) == 2 + rescuer = host.start(MEMBER + "alice") + host.flush(rescuer) + assert host.collector.accepted[build_a.run] == set(build_a.appended) + _assert_own_credential_only(host) + + @pytest.mark.parametrize("refused", [OUTSIDER, EXPIRED]) def test_a_passed_over_orphan_is_offered_again_once_the_login_changes(host, refused): """Once a service's own run is local-only and every other pending run is @@ -893,6 +919,26 @@ def test_a_lease_is_exclusive_until_released(tmp_path): taken.release() +def test_the_lease_files_keep_the_layout_other_services_expect(tmp_path): + """PolicyEngine/microcosm-emitter's telemetry service can share this spool + and keeps the same lease files: + ``.leases/.lock``. Any change here + makes services from different versions of either package blind to each + other's live runs.""" + + leases = ProducerLeases.beside(tmp_path / "events.sqlite3") + + assert leases.directory == tmp_path / "events.sqlite3.leases" + assert leases.path("run-a", "producer-a") == leases.directory / ( + "ba5dc36777c5d2275b9015aeea8bf76f7fcdbb2fc890d91fd89f5243836366f9.lock" + ) + held = leases.hold("run-a", "producer-a", timeout_seconds=1) + assert held is not None + assert oct(leases.directory.stat().st_mode & 0o777) == oct(0o700) + assert oct(held.path.stat().st_mode & 0o777) == oct(0o600) + held.release() + + def test_sweep_removes_only_free_leases_of_runs_with_nothing_pending(tmp_path): leases = ProducerLeases(tmp_path / "leases") live = leases.hold("run-live", "p", timeout_seconds=1)