Make a settled collector rejection local-only instead of retrying it forever

README.md

@@ -16,7 +16,7 @@ The service interprets requests such as a stage change or build failure, timesta
 
 The adapter bounds socket calls and reports telemetry failures without failing the build. It converts Python arguments into JSON but does not process their telemetry meaning. Network requests run on the service worker, outside the message-handling lock.
 
-The service uses the existing Hugging Face credential from the environment or local login cache. The collector validates PolicyEngine membership before issuing a short-lived write credential, and the first Hugging Face user to register a run owns it. Missing or rejected credentials permanently mark the service's own producer's events as local-only. Transient collector failures retain events for retry. The production destination belongs to the package; development overrides accept loopback addresses only.
+The service uses the existing Hugging Face credential from the environment or local login cache. The collector validates PolicyEngine membership before issuing a short-lived write credential, and the first Hugging Face user to register a run owns it. Missing or rejected credentials permanently mark the service's own producer's events as local-only. Transient collector failures (a timeout, a rate limit, a server error) retain events for retry. A refusal of a run's registration or events that is about the run and not the login (a 4xx other than 401, 403, 408 and 429) makes that run local-only instead, whichever service sends it. The production destination belongs to the package; development overrides accept loopback addresses only.
 
 Builds that supply the same queue path share one queue, so it also holds other builds' runs. A service's login decides only its own run. Each service holds an advisory lock file beside the queue, in `<queue>.leases/`, for its whole life. Another service delivers that run's remaining events only once the lock is free, which means the producing service has exited. It does so with its own login, and only while the collector accepts that login for the run; otherwise the run stays pending for another service. A run with no lock file, from a service that predates them, is left alone until it has been idle for 15 minutes. Microcosm's telemetry service keeps the same lock files from PolicyEngine/microcosm#1177 on, so services from both packages recognise each other's runs. The shutdown drain waits only for the service's own run.
 

changelog.d/telemetry-settled-rejection-local-only.fixed.md

@@ -0,0 +1 @@
+The telemetry service no longer retries a settled collector rejection forever. A 4xx other than a timeout (408) or a rate limit (429) on a run's events or registration (an event shape the collector does not accept, say) now makes that run local-only with a recorded reason and one warning. For a refused batch the reason is `collector_rejected_events` and the warning names the status; a refused registration keeps the reason and the warning a 409 already had. Before, the service sent the same request again without end, and the retry delay it set held back every other run it delivers. The rule is about the run and not about who sends it, so a service that has taken over another build's run applies it there too, with no warning in its own build. An expired collector token (401) is still exchanged again, a refused login (403) is handled as before, and a timeout, a rate limit or a server error is still retried. This ports PolicyEngine/microcosm#1147's rule, so both packages' services treat a shared queue's runs alike.

src/microcosm_emitter/telemetry/service/collector.py

@@ -31,6 +31,7 @@
     LOCAL_ONLY_MISSING_CREDENTIAL,
     LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION,
     LOCAL_ONLY_REJECTED_CREDENTIAL,
+    LOCAL_ONLY_REJECTED_EVENTS,
     LOCAL_ONLY_REJECTED_REGISTRATION,
     LOOPBACK_HOSTS,
     MAX_HTTP_RESPONSE_BYTES,
@@ -41,6 +42,7 @@
     OWN_LEASE_TIMEOUT_SECONDS,
     PRODUCTION_COLLECTOR_URL,
     REJECTED_CREDENTIAL_MESSAGE,
+    REJECTED_EVENTS_MESSAGE,
     RUN_EVENTS_PATH_TEMPLATE,
     RUN_REGISTRATION_PATH,
     TOKEN_EXCHANGE_PATH,
@@ -134,6 +136,18 @@ def _huggingface_token() -> str | None:
     )
 
 
+#: Client errors worth retrying: a timeout and a rate limit pass with time.
+#: Every other 4xx is the collector's settled answer to the same request, so
+#: retrying it only wedges the queue (a payload shape it does not accept, say).
+_RETRYABLE_CLIENT_ERRORS = frozenset(
+    {HTTPStatus.REQUEST_TIMEOUT, HTTPStatus.TOO_MANY_REQUESTS}
+)
+
+
+def _permanent_client_error(status: int) -> bool:
+    return 400 <= int(status) < 500 and status not in _RETRYABLE_CLIENT_ERRORS
+
+
 def _run(registration: Mapping[str, Any]) -> tuple[str, str]:
     """Identify a producer's run. Ids may contain any separator, so the pair
     itself is the key: joined into one string, two runs could share it."""
@@ -358,6 +372,13 @@ def _deliver(self, registration: Mapping[str, Any]) -> bool:
                 LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION,
                 caller_dependent=True,
             )
+        elif _permanent_client_error(status):
+            self._make_local_only(
+                registration,
+                LOCAL_ONLY_REJECTED_EVENTS,
+                caller_dependent=False,
+                message=REJECTED_EVENTS_MESSAGE.format(status=int(status)),
+            )
         else:
             self._defer_retry()
         return False
@@ -440,9 +461,9 @@ def _register(self, registration: Mapping[str, Any], token: str) -> bool:
                 LOCAL_ONLY_REJECTED_REGISTRATION,
                 caller_dependent=True,
             )
-        elif status == HTTPStatus.CONFLICT:
+        elif status == HTTPStatus.CONFLICT or _permanent_client_error(status):
             # A conflict reaches only the run's owner, so it is about the
-            # registration itself.
+            # registration itself, as is any other settled 4xx.
             self._make_local_only(
                 registration,
                 LOCAL_ONLY_REJECTED_REGISTRATION,
@@ -458,6 +479,7 @@ def _make_local_only(
         reason: str,
         *,
         caller_dependent: bool,
+        message: str = REJECTED_CREDENTIAL_MESSAGE,
     ) -> None:
         """Stop uploading a run, unless the reason is only this service's.
 
@@ -475,17 +497,17 @@ def _make_local_only(
                 self._passed_over[run] = self._credential
             return
         self.spool.make_local_only(*run, reason)
-        # The warning says "this run": another build's run is not this build's
+        # The warnings say "this run": another build's run is not this build's
         # to report, and its reason is kept in the spool.
         if own and reason != LOCAL_ONLY_MISSING_CREDENTIAL:
-            self._warn_denied(run)
+            self._warn_denied(run, message)
 
     def _defer_retry(self) -> None:
         self._next_attempt_at = time.monotonic() + self._retry_seconds
         self._retry_seconds = min(MAX_RETRY_SECONDS, self._retry_seconds * 2)
 
-    def _warn_denied(self, run: tuple[str, str]) -> None:
+    def _warn_denied(self, run: tuple[str, str], message: str) -> None:
         if run in self._warned_denied:
             return
-        print(REJECTED_CREDENTIAL_MESSAGE, file=sys.stderr, flush=True)
+        print(message, file=sys.stderr, flush=True)
         self._warned_denied.add(run)

src/microcosm_emitter/telemetry/service/constants.py

@@ -25,6 +25,7 @@
 LOCAL_ONLY_REJECTED_CREDENTIAL: Final = "huggingface_credential_rejected"
 LOCAL_ONLY_REJECTED_REGISTRATION: Final = "run_registration_rejected"
 LOCAL_ONLY_REJECTED_COLLECTOR_AUTHORIZATION: Final = "collector_authorization_rejected"
+LOCAL_ONLY_REJECTED_EVENTS: Final = "collector_rejected_events"
 LOCAL_ONLY_PRE_ELIGIBILITY: Final = "created_before_upload_eligibility"
 
 NO_CREDENTIAL_MESSAGE: Final = (
@@ -36,6 +37,11 @@
     "credential was not accepted as a PolicyEngine organization member. The "
     "dataset build will continue."
 )
+REJECTED_EVENTS_MESSAGE: Final = (
+    "Microcosm telemetry is local-only for this run: the collector rejected its "
+    "events (HTTP {status}), so they will not be retried. The dataset build will "
+    "continue."
+)
 OWN_LEASE_UNAVAILABLE_MESSAGE: Final = (
     "Microcosm telemetry is off for this run: its service could not take the "
     "lock that marks the run as live, and without it another build's service "