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..87adb7e43c --- /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 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 0d7e4d294a..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 @@ -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,11 @@ TOKEN_EXCHANGE_PATH, TOKEN_REFRESH_MARGIN_SECONDS, ) +from microcosm.build.telemetry_emitter_service.leases import ( + OwnLeaseUnavailableError, + ProducerLease, + ProducerLeases, +) from microcosm.build.telemetry_emitter_service.spool import EventSpool @@ -137,18 +148,39 @@ 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"]) + + +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 +188,52 @@ 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._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 + # 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[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.""" + + 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,60 +241,174 @@ 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: + pending = self.spool.pending_runs() + pending_runs = {_run(registration) for registration in pending} + self._passed_over = { + 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) + if claim is None: continue - if registration_key not in self._registered: - if not self._register(registration, token): + 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 - 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() - 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() + if self._deliver(registration): + made_progress = True return made_progress - def _collector_token(self, registration: Mapping[str, Any]) -> str | None: + 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() + 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 + 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 _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: + 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, producer_id = _run(registration) + token = self._collector_token(registration) + if token is None: + return False + if (run_id, producer_id) not in self._registered: + if not self._register(registration, token): + return False + events = self.spool.batch(run_id, producer_id) + 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: + session_token = self._cached_session_token() + if session_token is not None: + return session_token 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,13 +428,17 @@ 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 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, @@ -259,15 +449,26 @@ 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: 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,21 +478,36 @@ 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"]) - registration_key = _registration_key(registration) - self.spool.make_local_only(run_id, producer_id, reason) - if reason != LOCAL_ONLY_MISSING_CREDENTIAL: - self._warn_denied(registration_key, message) + """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 = _run(registration) + own = run == self._own_run + if caller_dependent and not own: + if reason != LOCAL_ONLY_MISSING_CREDENTIAL: + self._passed_over[run] = self._credential + return + 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(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 83163fe53c..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" @@ -60,6 +65,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..6fe562b7e7 --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/telemetry_emitter_service/leases.py @@ -0,0 +1,226 @@ +"""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 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.""" + + 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 and errors. + + Only a sweeper or an adopter holds another producer's lease, and only + 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: + lease = 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 + 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) + 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..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 @@ -38,16 +41,28 @@ 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. 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, 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..7ba5224d84 --- /dev/null +++ b/packages/microcosm-build/tests/engine_free/shared/test_telemetry_foreign_runs.py @@ -0,0 +1,1330 @@ +"""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 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 + 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 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, 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 +import sys +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 +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, example, 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 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 ( + OwnLeaseUnavailableError, + 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() + + +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, + conflicted_run_ids: set[str] | None = None, + ) -> None: + self.rejected_run_ids = set(rejected_run_ids or ()) + self.owners: dict[str, str] = {} + # A conflicted run id was registered before with other metadata, so + # whoever owns it gets 409 for this registration. + 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, 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 + during() + path = urlsplit(url).path + 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( + 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 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 + 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 _run_of(self.registration) + + +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, + # 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]] = [] + # 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}", 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( + 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, + len(self.collector.requests), + ) + ) + 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 + + 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() + + 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 + 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 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: + # 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 latest in (("register", run, 403), ("register", run, 409)), ( + run, + latest, + ) + elif reason == LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION: + assert latest == ("events", run, 403), (run, latest) + else: + assert reason == LOCAL_ONLY_REJECTED_EVENTS + assert latest in (("events", run, 404), ("events", run, 422)), ( + run, + latest, + ) + # 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) + + +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") + 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) + + +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 + 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 +): + 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_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) + 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_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( + """ + 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 _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 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]]] = {} + 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.") + # B's first delivery pass, which meets A's queued run too, makes B's own + # 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" + time.sleep(0.1) + time.sleep(1.5) + 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() + + with closing(sqlite3.connect(spool_path)) as connection: + states = dict( + connection.execute( + "SELECT run_id, COALESCE(local_only_reason, upload_state) " + "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: + """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)), + # 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, +) + + +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), + 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 +# flight, and A then uploaded B's run. +@example( + 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, conflicted, rescue_logins +): + host = Host( + tmp_path_factory.mktemp("foreign"), + FakeCollector( + {f"run-{index}" for index in rejected}, + {f"run-{index}" for index in conflicted}, + ), + ) + 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) + 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) + + # 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) + 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 + 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", + 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()