From adcb4704d47d0485426fe8d51b49b108a8335754 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 21:58:25 -0700 Subject: [PATCH 1/4] =?UTF-8?q?refactor(hermes=5Fcli):=20relay=5Fshared=5F?= =?UTF-8?q?metrics=20=E2=80=94=20=5Fforget()=20for=20owner-checked=20index?= =?UTF-8?q?=20eviction?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../observability/relay_shared_metrics.py | 39 +++++++++---------- 1 file changed, 18 insertions(+), 21 deletions(-) diff --git a/hermes_cli/observability/relay_shared_metrics.py b/hermes_cli/observability/relay_shared_metrics.py index a382bf18e9..721d034124 100644 --- a/hermes_cli/observability/relay_shared_metrics.py +++ b/hermes_cli/observability/relay_shared_metrics.py @@ -66,6 +66,12 @@ def _retry_ordinal(event: dict[str, Any]) -> int | None: return None +def _forget(index: dict[Any, _MetricsSession], key: Any, owner: _MetricsSession) -> None: + """Drop ``key`` from ``index`` only while it still points at ``owner``.""" + if index.get(key) is owner: + index.pop(key, None) + + def _sole(items: Any) -> Any: """The single distinct element of ``items`` (identity-deduplicated), else None.""" unique = {id(item): item for item in items} @@ -469,8 +475,7 @@ class _Runtime: else: self._export() with self._sessions_lock: - if self._sessions.get(session.session_id) is session: - self._sessions.pop(session.session_id, None) + _forget(self._sessions, session.session_id, session) def shutdown(self) -> None: with self._sessions_lock: @@ -519,12 +524,11 @@ class _Runtime: """ with self._send_lock: thread = self._send_thread - if thread is None or not thread.is_alive(): - return - try: - thread.join(timeout) - except Exception: - logger.debug("Shared-metrics send thread join failed", exc_info=True) + if thread is not None and thread.is_alive(): + try: + thread.join(timeout) + except Exception: + logger.debug("Shared-metrics send thread join failed", exc_info=True) def _session(self, event: dict[str, Any]) -> _MetricsSession | None: with self._sessions_lock: @@ -790,13 +794,9 @@ class _Runtime: session.tasks.pop(task_id, None) session.retired_turn_ids.extend(task.turn_ids) with self._task_sessions_lock: - task_key = (session.session_id, task_id) - if self._task_sessions.get(task_key) is session: - self._task_sessions.pop(task_key, None) + _forget(self._task_sessions, (session.session_id, task_id), session) for turn_id in task.turn_ids: - turn_key = (session.session_id, turn_id) - if self._turn_sessions.get(turn_key) is session: - self._turn_sessions.pop(turn_key, None) + _forget(self._turn_sessions, (session.session_id, turn_id), session) return True def _export(self) -> None: @@ -831,14 +831,11 @@ class _Runtime: # One in-flight pass per process; the next hook fire picks up what is pending. if self._send_thread is not None and self._send_thread.is_alive(): return - thread = threading.Thread( - target=self._run_send_pass, - args=(resolved.endpoint,), - name="hermes-shared-metrics-send", - daemon=True, + self._send_thread = threading.Thread( + target=self._run_send_pass, args=(resolved.endpoint,), + name="hermes-shared-metrics-send", daemon=True, ) - self._send_thread = thread - thread.start() + self._send_thread.start() def _run_send_pass(self, endpoint: str) -> None: from hermes_cli.observability.shared_metrics_sender import SharedMetricsSender From 9cda9c1f97931601d7b0f9b92021caab6bba3f29 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:19:53 -0700 Subject: [PATCH 2/4] =?UTF-8?q?refactor(hermes=5Fcli):=20shared=5Fmetrics?= =?UTF-8?q?=5Fcontract=20=E2=80=94=20kwargs-driven=20shape=20matcher,=20re?= =?UTF-8?q?flowed=20allowlist=20constants=20(values=20unchanged)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../observability/shared_metrics_contract.py | 123 ++++++------------ 1 file changed, 40 insertions(+), 83 deletions(-) diff --git a/hermes_cli/observability/shared_metrics_contract.py b/hermes_cli/observability/shared_metrics_contract.py index cefab49d6c..344b17546a 100644 --- a/hermes_cli/observability/shared_metrics_contract.py +++ b/hermes_cli/observability/shared_metrics_contract.py @@ -37,63 +37,49 @@ PROVIDER_IDENTIFIER_MAX_LENGTH = 64 _METRIC_IDENTIFIER_CHARACTERS = frozenset("abcdefghijklmnopqrstuvwxyz0123456789._:/@+-") _METRIC_IDENTIFIER_START_CHARACTERS = frozenset("abcdefghijklmnopqrstuvwxyz0123456789") -EXECUTION_SURFACES: frozenset[str] = frozenset({ - "api", "batch", "cli", "desktop", "gateway", "python", "scheduled_task", "tui", - "other", "unknown", +EXECUTION_SURFACES = frozenset({ + "api", "batch", "cli", "desktop", "gateway", "python", "scheduled_task", "tui", "other", + "unknown", }) -TASK_OUTCOMES: frozenset[str] = frozenset({ - "cancelled", "failed", "success", "timed_out", "unknown", -}) -TASK_END_REASONS: frozenset[str] = frozenset({ +TASK_OUTCOMES = frozenset({"cancelled", "failed", "success", "timed_out", "unknown"}) +TASK_END_REASONS = frozenset({ "approval_denied", "completed", "failed", "guardrail_blocked", "iteration_limit", "system_aborted", "timed_out", "unknown", "user_cancelled", }) -TASK_TERMINATIONS: frozenset[str] = frozenset({ - "none", "system_aborted", "timed_out", "unknown", "user_cancelled", +TASK_TERMINATIONS = frozenset({"none", "system_aborted", "timed_out", "unknown", "user_cancelled"}) +TASK_ENTRYPOINTS = frozenset({ + "api", "background", "batch", "delegated", "gateway_message", "interactive", "other", "python", + "scheduled_task", "unknown", }) -TASK_ENTRYPOINTS: frozenset[str] = frozenset({ - "api", "background", "batch", "delegated", "gateway_message", "interactive", "other", - "python", "scheduled_task", "unknown", -}) -DURATION_BUCKETS: frozenset[str] = frozenset({ +DURATION_BUCKETS = frozenset({ "1s_to_5s", "2m_to_10m", "30s_to_2m", "5s_to_30s", "gte_10m", "lt_1s", }) -COUNT_BUCKETS: frozenset[str] = frozenset({"0", "1", "2", "3_to_5", "6_to_10", "gte_11"}) -TOOL_CATEGORIES: frozenset[str] = frozenset({ +COUNT_BUCKETS = frozenset({"0", "1", "2", "3_to_5", "6_to_10", "gte_11"}) +TOOL_CATEGORIES = frozenset({ "browser", "code_execution", "communication", "computer_use", "delegation", "file", - "home_automation", "mcp", "media", "memory", "other", "planning", "project", - "scheduler", "skill", "terminal", "unknown", "web", + "home_automation", "mcp", "media", "memory", "other", "planning", "project", "scheduler", + "skill", "terminal", "unknown", "web", }) -TOOL_OUTCOMES: frozenset[str] = frozenset({ - "blocked", "cancelled", "failed", "success", "timed_out", "unknown", -}) -TOOL_APPROVAL_OUTCOMES: frozenset[str] = frozenset({ - "approved", "denied", "not_required", "timed_out", "unknown", -}) -TOOL_APPROVAL_ATTRIBUTIONS: frozenset[str] = frozenset({"tool_call", "unattributed"}) -TOOL_LATENCY_BUCKETS: frozenset[str] = frozenset({ +TOOL_OUTCOMES = frozenset({"blocked", "cancelled", "failed", "success", "timed_out", "unknown"}) +TOOL_APPROVAL_OUTCOMES = frozenset({"approved", "denied", "not_required", "timed_out", "unknown"}) +TOOL_APPROVAL_ATTRIBUTIONS = frozenset({"tool_call", "unattributed"}) +TOOL_LATENCY_BUCKETS = frozenset({ "100ms_to_250ms", "10s_to_30s", "1s_to_2s", "250ms_to_500ms", "2s_to_5s", "500ms_to_1s", "5s_to_10s", "gte_30s", "lt_100ms", "unknown", }) -TOOL_RETRY_BUCKETS: frozenset[str] = COUNT_BUCKETS | frozenset({"unknown"}) -SKILL_LIFECYCLE_ACTIONS: frozenset[str] = frozenset({ +TOOL_RETRY_BUCKETS = COUNT_BUCKETS | frozenset({"unknown"}) +SKILL_LIFECYCLE_ACTIONS = frozenset({ "archived", "created", "edited", "installed", "patched", "restored", "stale", }) -SKILL_PROVENANCES: frozenset[str] = frozenset({ - "agent_created", "external", "installed", "local", "unknown", -}) -SKILL_REUSE_STATES: frozenset[str] = frozenset({"first_use", "reused"}) -SKILL_POST_PATCH_STATES: frozenset[str] = frozenset({ - "no_new_patch", "not_applicable", "reused_after_patch", -}) -CLIENT_OS_FAMILIES: frozenset[str] = frozenset({"linux", "macos", "unknown", "windows"}) -CLIENT_ARCHITECTURES: frozenset[str] = frozenset({"arm", "arm64", "unknown", "x86", "x86_64"}) -CLIENT_INSTALL_METHODS: frozenset[str] = frozenset({ +SKILL_PROVENANCES = frozenset({"agent_created", "external", "installed", "local", "unknown"}) +SKILL_REUSE_STATES = frozenset({"first_use", "reused"}) +SKILL_POST_PATCH_STATES = frozenset({"no_new_patch", "not_applicable", "reused_after_patch"}) +CLIENT_OS_FAMILIES = frozenset({"linux", "macos", "unknown", "windows"}) +CLIENT_ARCHITECTURES = frozenset({"arm", "arm64", "unknown", "x86", "x86_64"}) +CLIENT_INSTALL_METHODS = frozenset({ "apt", "docker", "git", "home-manager", "homebrew", "nixos", "pip", "unknown", }) -CLIENT_RESOURCE_KEYS: frozenset[str] = frozenset({ - "architecture", "hermes_version", "install_method", "os_family", -}) +CLIENT_RESOURCE_KEYS = frozenset({"architecture", "hermes_version", "install_method", "os_family"}) _ARCHITECTURE_ALIASES = { "amd64": "x86_64", "x64": "x86_64", "x86_64": "x86_64", @@ -134,12 +120,10 @@ def client_resource( hermes_version: Any, *, os_name: Any, architecture: Any, install_method: Any ) -> dict[str, str]: """Build the bounded client resource attached to aggregate packages.""" - normalized_version = str(hermes_version or "").strip() - if not normalized_version or len(normalized_version) > 64: - normalized_version = "unknown" + version = str(hermes_version or "").strip() return { "architecture": client_architecture(architecture), - "hermes_version": normalized_version, + "hermes_version": version if 0 < len(version) <= 64 else "unknown", "install_method": client_install_method(install_method), "os_family": client_os_family(os_name), } @@ -223,7 +207,7 @@ _METRIC_FIELDS: dict[str, frozenset[str]] = { **{name: frozenset(contract) for name, contract in _COUNTER_DIMENSION_VALUES.items()}, MODEL_ROUTE_METRIC: frozenset(_MODEL_ROUTE_MAX_LENGTHS), } -COUNTER_METRICS: frozenset[str] = frozenset(_METRIC_FIELDS) - {LEGACY_MODEL_CALL_METRIC} +COUNTER_METRICS = frozenset(_METRIC_FIELDS) - {LEGACY_MODEL_CALL_METRIC} _SKILL_MARK_METRICS = { SKILL_LIFECYCLE_MARK: SKILL_LIFECYCLE_METRIC, SKILL_LOAD_MARK: SKILL_LOAD_METRIC, @@ -246,9 +230,7 @@ def counter_dimensions_are_valid(metric_name: str, dimensions: dict[str, Any]) - ) -def _relay_metadata( - event: Any, schema_key: str, schema_version: str, *extra_keys: str -) -> dict | None: +def _relay_metadata(event: Any, schema_key: str, schema_version: str, *extra_keys: str) -> dict | None: """Return the event metadata when it carries only the allowlisted Relay keys.""" metadata = getattr(event, "metadata", None) if not isinstance(metadata, dict) or metadata.get(schema_key) != schema_version: @@ -259,44 +241,21 @@ def _relay_metadata( return metadata -_ANY = object() - - def _event_text(event: Any, attr: str) -> str: return str(getattr(event, attr, "") or "") -def _event_shape_matches( - event: Any, - *, - kind: str, - name: Any = _ANY, - category: Any = _ANY, - scope_category: Any = _ANY, - category_profile: Any = _ANY, -) -> bool: - """Match the coarse Relay event shape. +def _event_shape_matches(event: Any, **expected: Any) -> bool: + """Match the coarse Relay event shape (``kind`` plus any of name/category/scope_category/ + category_profile). A ``str`` expectation compares against the stringified attribute; ``None`` requires the - attribute itself to be ``None``; ``_ANY`` skips the check. Anything else (the - ``category_profile`` dict) compares with plain equality. + attribute itself to be ``None``; anything else (the ``category_profile`` dict) compares + with plain equality. Unmentioned attributes are not checked. """ - if _event_text(event, "kind") != kind: - return False - expectations = { - "name": name, "category": category, "scope_category": scope_category, - "category_profile": category_profile, - } - for attr, expected in expectations.items(): - if expected is _ANY: - continue - if isinstance(expected, str): - matched = _event_text(event, attr) == expected - elif expected is None: - matched = getattr(event, attr, None) is None - else: - matched = getattr(event, attr, None) == expected - if not matched: + for attr, value in expected.items(): + actual = _event_text(event, attr) if isinstance(value, str) else getattr(event, attr, None) + if actual != value: return False return True @@ -470,9 +429,7 @@ def execution_surface(kwargs: dict[str, Any]) -> str: return "gateway" except Exception: pass - if value in _KNOWN_GATEWAY_PLATFORMS: - return "gateway" - return "unknown" if value == "unknown" else "other" + return "gateway" if value in _KNOWN_GATEWAY_PLATFORMS else "other" def task_start_fields(kwargs: dict[str, Any]) -> dict[str, str]: From 17650193b7d4eb25769b8f919aa78363e3aacb8c Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:23:49 -0700 Subject: [PATCH 3/4] =?UTF-8?q?refactor(hermes=5Fcli):=20relay=5Fshared=5F?= =?UTF-8?q?metrics=20=E2=80=94=20drop=20write-only=20=5FModelCall.retry=5F?= =?UTF-8?q?ordinal=20and=20unread=20=5FToolCall.task=5Fid?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hermes_cli/observability/relay_shared_metrics.py | 10 ++-------- 1 file changed, 2 insertions(+), 8 deletions(-) diff --git a/hermes_cli/observability/relay_shared_metrics.py b/hermes_cli/observability/relay_shared_metrics.py index 721d034124..568cbe38c1 100644 --- a/hermes_cli/observability/relay_shared_metrics.py +++ b/hermes_cli/observability/relay_shared_metrics.py @@ -83,13 +83,11 @@ class _ModelCall: handle: Any task_id: str fields: dict[str, str] - retry_ordinal: int | None = None @dataclass class _ToolCall: handle: Any - task_id: str category: str started_ns: int approval_outcome: str = "not_required" @@ -293,8 +291,6 @@ class _Runtime: # attempt. Provider fallback resets Hermes's provider-local retry # ordinal, so ordinal deltas are not a reliable task-level counter. task.retry_count += 1 - if retry_ordinal is not None: - existing.retry_ordinal = max(existing.retry_ordinal or 0, retry_ordinal) return if task is not None: task.model_call_ids.add(request_id) @@ -308,9 +304,7 @@ class _Runtime: metadata=self._event_metadata(), model_name=MODEL_CALL_PROFILE_MODEL, ) - session.model_calls[model_call_key] = _ModelCall( - handle=handle, task_id=task_id, fields=fields, retry_ordinal=retry_ordinal - ) + session.model_calls[model_call_key] = _ModelCall(handle, task_id, fields) def record_model_call_error(self, event: dict[str, Any]) -> None: """Retain the latest attempt error without closing the logical call.""" @@ -713,7 +707,7 @@ class _Runtime: task, self.relay.tools.call, TOOL_CALL_SCOPE, {}, handle=task.handle, metadata=self._event_metadata(), ) - return _ToolCall(handle, task.task_id, tool_category(event), monotonic_ns()) + return _ToolCall(handle, tool_category(event), monotonic_ns()) def _finish_tool_call( self, task: _TaskRun, tool_call: _ToolCall, event: dict[str, Any] From ad079a5f43157e9af45154e0fd70fff374569e66 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:27:25 -0700 Subject: [PATCH 4/4] =?UTF-8?q?refactor(hermes=5Fcli):=20shared=5Fmetrics?= =?UTF-8?q?=5Fsender=20=E2=80=94=20dataclass=20=5FResponse,=20compact=20ra?= =?UTF-8?q?tionale=20comments=20(invariants=20kept)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../observability/shared_metrics_sender.py | 150 +++++++----------- 1 file changed, 56 insertions(+), 94 deletions(-) diff --git a/hermes_cli/observability/shared_metrics_sender.py b/hermes_cli/observability/shared_metrics_sender.py index dab55fdbd1..530d9b2c3d 100644 --- a/hermes_cli/observability/shared_metrics_sender.py +++ b/hermes_cli/observability/shared_metrics_sender.py @@ -41,35 +41,29 @@ GZIP_THRESHOLD_BYTES = 4096 #: Packages per pass. Bounds work on an interactive hook even after an outage. MAX_PACKAGES_PER_PASS = 20 -#: How long a claimed row is held. The claim writes a LEASE INTO THE FUTURE: -#: selection requires `next_attempt_at <= now`, so no other process can take the -#: package for the length of the lease. Must exceed the worst case for ONE package -#: (three 30s timeouts plus 1s+5s backoff, ~96s) — which is why packages are claimed -#: one at a time right before sending: a batch of 20 under one lease can legally run -#: ~1900s, so later rows' leases expired while still held and got re-sent. +#: Claim lease, written INTO THE FUTURE as next_attempt_at so no other process selects +#: the row meanwhile. Must exceed one package's worst case (3x30s timeouts + 1s+5s +#: backoff, ~96s) — hence packages are claimed one at a time right before sending; a +#: batch of 20 under one lease can run ~1900s and get re-sent by another process. _CLAIM_LEASE_SECONDS = 300 #: Floor applied after a pass fails to deliver, so a hard-down service is not #: retried on every task completion. _FAILURE_BACKOFF_SECONDS = 15 * 60 -#: Permanent statuses per the ingest contract. Deliberately narrow: 400 means the -#: envelope will never validate; 413 because a package over the 1 MiB cap cannot -#: shrink on retry. Everything else — including 403 from the origin guard and 404 -#: from a bad path — is retried, since those are usually deployment/edge -#: misconfiguration that resolves without the package changing. +#: Permanent statuses per the ingest contract, deliberately narrow: 400 never validates, +#: 413 (over the 1 MiB cap) cannot shrink on retry. Everything else — including 403 from +#: the origin guard and 404 from a bad path — is deployment/edge misconfiguration that +#: resolves without the package changing, so it is retried. _PERMANENT_STATUSES = frozenset({400, 413}) -#: Attempts after which a package is abandoned. Without a ceiling a poisoned row -#: is retried until 30-day retention deletes it (~160 requests), wasting the user's -#: bandwidth and pinning a doomed package at the head of the queue. +#: Attempts after which a package is abandoned; otherwise a poisoned row is retried until +#: 30-day retention deletes it (~160 requests) and pins the head of the queue. MAX_SEND_ATTEMPTS = 25 -#: Maximum distance one reconcile call can advance the 'obs' mark. Honest heartbeats -#: arrive hours apart so the cap never binds normally; a machine off for months catches -#: up in a few hook fires (fail-closed latency only). It bounds FORWARD clock poison: -#: without it one glitched sample (NTP flap reading 2099) drags the mark — and every -#: window open and confirmation horizon — decades ahead, a refused-data leak. +#: Max distance one reconcile can advance the 'obs' mark. Never binds for honest +#: heartbeats (hours apart); bounds FORWARD clock poison, where one glitched sample (NTP +#: flap reading 2099) would drag every window open and confirmation horizon decades ahead. MAX_OBS_ADVANCE_SECONDS = 30 * 24 * 3600 @@ -87,13 +81,11 @@ class SendOutcome: deferred: int = 0 +@dataclass(slots=True) class _Response: - __slots__ = ("status", "retry_after", "body") - - def __init__(self, status: int, retry_after: str | None, body: str) -> None: - self.status = status - self.retry_after = retry_after - self.body = body + status: int + retry_after: str | None + body: str def _post(endpoint: str, payload: bytes, *, timeout: int) -> _Response: @@ -144,8 +136,8 @@ def reconcile_send_consent( ) -> None: """Reconcile the consent-window table with the observed config state. - THE ONLY writer of consent state. Must run inside a write transaction. A pure function - of (config, now, store): call it from anywhere, any number of times, in any order. + THE ONLY writer of consent state; must run inside a write transaction. Idempotent in + (config, now, store): call it from anywhere, any number of times, in any order. """ stamp = _isoformat(now or _utc_now()) raw_stamp = stamp # pre-cap observation time, used to clamp closes @@ -188,13 +180,11 @@ def reconcile_send_consent( (obs, open_row[0]), ) elif open_row is not None: - # Close at the last CONFIRMED moment, but never after the closing observation's - # own raw stamp. Both clamps are load-bearing: min with last_confirmed_at means an - # unobserved gap (machine off, hand-edited config) is never asserted as consented; - # min with the RAW (pre-cap, pre-MAX) stamp means a last_confirmed_at poisoned by a - # glitched-forward sample is pulled back to the true revoke moment by an honest - # clock, so the refused era falls OUTSIDE the window. A rolled-back clock only - # closes EARLIER — fail-closed. + # Close at the last CONFIRMED moment, never after the closing observation's RAW + # stamp. Both clamps are load-bearing: last_confirmed_at means an unobserved gap + # (machine off, hand-edited config) is never asserted as consented; the raw + # (pre-cap) stamp pulls a glitched-forward last_confirmed_at back to the true + # revoke moment. A rolled-back clock only closes EARLIER — fail-closed. connection.execute( "UPDATE send_consent_windows" " SET closed_at = MIN(last_confirmed_at, ?)" @@ -204,9 +194,8 @@ def reconcile_send_consent( #: Claim-time consent predicate: the package's period must fall entirely inside SOME -#: recorded consent window. An open window vouches only up to its last confirmed moment, -#: so a package whose period runs past it waits for the next reconcile heartbeat -#: (fail-closed; released within one hook fire). +#: recorded window. An open window vouches only up to its last confirmed moment, so a +#: package running past it waits for the next reconcile heartbeat (fail-closed). CONSENT_GATE_SQL = """EXISTS ( SELECT 1 FROM send_consent_windows w WHERE package_outbox.period_start >= w.opened_at @@ -250,13 +239,11 @@ class SharedMetricsSender: # -- selection --------------------------------------------------------- def _claim_next(self, now: datetime, seen: set[str]) -> dict | None: - """Claim exactly ONE package, immediately before it is sent. + """Claim exactly ONE package, immediately before it is sent (see _CLAIM_LEASE_SECONDS). - Claiming a batch up front fails: one shared lease would have to cover the whole - pass, and 20 retrying packages can outlive any sane lease. ``seen`` (packages this - pass finished with) is excluded IN SQL: with LIMIT 1, returning None for a seen row - would look like an empty queue and abandon everything behind it, and rows can - legitimately become eligible again mid-pass. + ``seen`` (packages this pass finished with) is excluded IN SQL: with LIMIT 1, + returning None for a seen row would look like an empty queue and abandon everything + behind it, and rows can legitimately become eligible again mid-pass. """ with self._write() as connection: stamp = _isoformat(now) @@ -264,8 +251,7 @@ class SharedMetricsSender: placeholders = ",".join("?" for _ in seen) exclusion = f" AND package_id NOT IN ({placeholders})" if seen else "" - # Consent is a READ here — the claim must never mutate the window table, or - # selecting a row could rewrite what is permitted to be sent. + # Consent is a READ here: the claim must never mutate the window table. row = connection.execute( f""" SELECT package_id, payload_json, sent_install_id @@ -300,10 +286,8 @@ class SharedMetricsSender: claim_token = ? WHERE package_id = ? """, - # Lease INTO THE FUTURE (see _CLAIM_LEASE_SECONDS). The token is this - # claim's identity: a reclaim after expiry mints a new one, and every later - # write by THIS claimant is compare-and-set against it, so a lapsed claimant - # that resumes cannot settle or transmit. + # The token is this claim's identity: a reclaim after expiry mints a new one + # and every later write by THIS claimant is compare-and-set against it. (_isoformat(lease_until), token, package_id), ) return { @@ -320,9 +304,8 @@ class SharedMetricsSender: ) -> str | None: """Record the transmitted install_id on the row, or reject an unusable one. - ``sent_install_id`` records exactly what the wire will carry. Rejecting here rather - than raising matters: an exception rolls back the claim transaction and blocks every - healthy package behind this one. + Rejecting rather than raising matters: an exception would roll back the claim + transaction and block every healthy package behind this one. """ install_id = None try: @@ -363,12 +346,8 @@ class SharedMetricsSender: @staticmethod def _body(payload_json: str, transmitted_id: str) -> bytes: - """Rebuild the exact bytes to send. - - Recomputed from the stored package (json.dumps with these options is deterministic) - with install_id taken from the frozen ``sent_install_id`` column, keeping "a resend - is byte-identical" anchored to one recorded value. - """ + """Rebuild the exact bytes to send: deterministic json.dumps of the stored package + with install_id from the frozen ``sent_install_id`` column (resends are byte-identical).""" payload = dict(json.loads(payload_json)) payload["install_id"] = transmitted_id return json.dumps(payload, indent=2, sort_keys=True).encode("utf-8") @@ -383,10 +362,9 @@ class SharedMetricsSender: ) -> None: """Write send state for one package. - Guarded on send_state so a pass whose lease lapsed cannot resurrect a row another - process already finished (overwriting 'sent' back to 'pending' would re-send). With - ``token`` the write is also compare-and-set on claim_token: a superseded claimant - writes zero rows and its settlement/backoff/error silently lose to the newer claim. + Guarded on send_state so a lapsed pass cannot resurrect a row another process already + finished. With ``token`` the write is also compare-and-set on claim_token: a + superseded claimant writes zero rows. """ assignments = ", ".join(f"{name} = ?" for name in columns) predicate = " AND (send_state IS NULL OR send_state = 'pending')" if only_if_pending else "" @@ -401,13 +379,12 @@ class SharedMetricsSender: ) def _renew_claim(self, package_id: str, token: str | None) -> bool: - """Atomically re-assert ownership and extend the lease. CAS, one row. + """Atomically re-assert ownership and extend the lease (CAS, one row; rowcount == 1 + is the only grant). - A read-only ownership check is not enough: a claimant whose lease expired while - suspended still has its token in the row if nobody reclaimed yet, and would POST - while another process legitimately reclaims (check-to-POST expiry race). Pushing a - fresh lease in the same CAS keeps the upcoming POST (30s, well under the 300s lease) - entirely inside renewed authority. rowcount == 1 is the only grant. + A read-only check is not enough: a claimant whose lease expired while suspended still + has its token in the row and would POST while another process reclaims. Renewing in + the same CAS keeps the 30s POST inside the 300s lease. """ if token is None: return False @@ -435,9 +412,7 @@ class SharedMetricsSender: def _defer( self, package_id: str, delay_seconds: int, reason: str, *, token: str | None = None ) -> None: - # Clamp to >= 1s: a past deadline would make the row instantly re-eligible and let - # a pass spin on it. Unreachable today (Retry-After is pre-clamped, other callers - # pass positive constants) — kept as a guard against a future caller. + # Clamp to >= 1s so a past deadline can never make the row instantly re-eligible. retry_at = self._now().timestamp() + max(1, int(delay_seconds)) self._mark( package_id, @@ -450,9 +425,8 @@ class SharedMetricsSender: def _send_one(self, package: dict) -> str: """Try one package. Returns 'sent', 'rejected', or 'deferred'. - Delivery is at-least-once. The pre-POST renewal plus token-fenced writes close the - claim->POST and settle-after-reclaim gaps, but a suspension landing MID-POST can - still duplicate: no client-side check can revoke a request in flight. + Delivery is at-least-once: renewal plus token-fenced writes close the claim->POST + and settle-after-reclaim gaps, but a suspension MID-POST can still duplicate. """ package_id = package["package_id"] token = package.get("claim_token") @@ -463,10 +437,8 @@ class SharedMetricsSender: return "deferred" for attempt in range(1, self._max_attempts + 1): - # Renew before EVERY external POST (CAS on token, pending, lease unexpired), so a - # suspended-then-resumed claimant whose lease lapsed yields even before anyone - # reclaims. The ingest key is minute-prefixed, so duplicates become distinct - # stored objects, not overwrites. + # Renew before EVERY external POST so a lapsed claimant yields even before anyone + # reclaims. The ingest key is minute-prefixed: duplicates are distinct objects. if not self._renew_claim(package_id, token): logger.info( "Shared-metrics claim on %s superseded or expired; yielding", package_id @@ -484,9 +456,6 @@ class SharedMetricsSender: ) return "sent" if response.status in _PERMANENT_STATUSES: - # Only contract-terminal statuses. 403 in particular is the origin guard - # during an edge misconfiguration — treating it as permanent would - # discard every package sent during the incident. logger.warning( "Telemetry package %s rejected with HTTP %s; not retrying", package_id, @@ -518,19 +487,15 @@ class SharedMetricsSender: # -- entry point ------------------------------------------------------- def send_pending(self) -> SendOutcome: - """Run one bounded pass. Never raises. - - Claims and sends ONE package at a time so each row's lease only covers its own - transmission, and re-checks consent before every send so revoking `send` mid-pass - stops the remaining packages. - """ + """Run one bounded pass, one claim+send at a time, re-checking consent before each + send so revoking `send` mid-pass stops the remainder. Never raises.""" outcome = SendOutcome() seen: set[str] = set() for _ in range(MAX_PACKAGES_PER_PASS): if not self._still_consented(): - # Stop transmitting and reconcile through the single consent writer so - # the window closes at its last confirmed moment. + # Reconcile through the single consent writer so the window closes at its + # last confirmed moment. logger.info("Shared-metrics sending disabled mid-pass; stopping") self._reconcile(send_enabled=False) break @@ -564,11 +529,8 @@ class SharedMetricsSender: logger.warning("Unable to reconcile shared-metrics consent", exc_info=True) def _still_consented(self) -> bool: - """Re-read profile-owned send consent. - - Consent is a boundary, not cached config: docs promise ``send: false`` stops - transmission immediately, and a pass can run for minutes. - """ + """Re-read profile-owned send consent: docs promise ``send: false`` stops transmission + immediately, and a pass can run for minutes.""" if self._consent_check is None: return True try: