From 846889066e47803677d7656dfe3724953af26dd6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pawe=C5=82=20Chojnacki?= Date: Mon, 13 Jul 2026 16:57:02 +0200 Subject: [PATCH 1/7] test(telemetry): deduplicate fork-cloned heartbeats --- tests/_telemetry_heartbeat.py | 43 ++++++++++++++ tests/test_telemetry.py | 51 ++--------------- .../test_the_test/test_telemetry_heartbeat.py | 56 +++++++++++++++++++ 3 files changed, 105 insertions(+), 45 deletions(-) create mode 100644 tests/_telemetry_heartbeat.py create mode 100644 tests/test_the_test/test_telemetry_heartbeat.py diff --git a/tests/_telemetry_heartbeat.py b/tests/_telemetry_heartbeat.py new file mode 100644 index 00000000000..7a98414671f --- /dev/null +++ b/tests/_telemetry_heartbeat.py @@ -0,0 +1,43 @@ +import itertools +from collections import defaultdict +from collections.abc import Iterable +from typing import Any + +from dateutil.parser import isoparse + + +def heartbeat_delays_by_runtime( + telemetry_data: Iterable[dict[str, Any]], +) -> tuple[dict[str, list[float]], dict[str, int]]: + """Return heartbeat delays and logical heartbeat counts grouped by runtime ID.""" + heartbeats_by_runtime: dict[str, dict[int, dict[str, Any]]] = defaultdict(dict) + + for data in telemetry_data: + content: dict[str, Any] = data["request"]["content"] + if content.get("request_type") != "app-heartbeat": + continue + + runtime_id: str = content["runtime_id"] + seq_id: int = content["seq_id"] + + # A retry or a fork clone can send the same logical heartbeat more than once. + # It must contribute only once to the runtime's observed cadence. + heartbeats_by_runtime[runtime_id].setdefault(seq_id, data) + + heartbeat_counts = {runtime_id: len(heartbeats) for runtime_id, heartbeats in heartbeats_by_runtime.items()} + delays_by_runtime: dict[str, list[float]] = {} + + for runtime_id, heartbeats_by_seq_id in heartbeats_by_runtime.items(): + if len(heartbeats_by_seq_id) <= 2: + continue + + heartbeats = sorted( + heartbeats_by_seq_id.values(), + key=lambda data: isoparse(data["request"]["timestamp_start"]), + ) + times = [isoparse(data["request"]["timestamp_start"]) for data in heartbeats] + delays_by_runtime[runtime_id] = [ + (current - previous).total_seconds() for previous, current in itertools.pairwise(times) + ] + + return delays_by_runtime, heartbeat_counts diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index 49b910822b0..488446c445e 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -9,6 +9,7 @@ from dateutil.parser import isoparse +from tests._telemetry_heartbeat import heartbeat_delays_by_runtime from utils import bug, context, features, interfaces, logger, rfc, scenarios, weblog from utils.interfaces._misc_validators import HeadersMatchValidator, HeadersPresenceValidator from utils.telemetry import get_lang_configs, load_telemetry_json @@ -312,59 +313,19 @@ def save_data(data: dict, container: dict): raise Exception("The following telemetry messages were not forwarded by the agent") @staticmethod - def _get_heartbeat_delays_by_runtime() -> dict: + def _get_heartbeat_delays_by_runtime() -> dict[str, list[float]]: """Returns a dict where : The key is the runtime id The value is a list of delay observed on this runtime id """ - telemetry_data = list(interfaces.library.get_telemetry_data()) - heartbeats_by_runtime = defaultdict(list) - - for data in telemetry_data: - if data["request"]["content"].get("request_type") == "app-heartbeat": - heartbeats_by_runtime[data["request"]["content"]["runtime_id"]].append(data) - - delays_by_runtime = {} - - # Measure only long-lived application runtimes, not the short-lived children the session-id - # tests spawn via /spawn_child (a forked child can emit under its parent's runtime_id before - # regenerating its own -- a sub-interval "duplicate" that flakes the check). Select by lifespan, - # not heartbeat count, so a slow-drifting worker with fewer heartbeats still gets measured. - def lifespan(heartbeats: list[Any]) -> float: - times = [isoparse(d["request"]["timestamp_start"]) for d in heartbeats] - return (max(times) - min(times)).total_seconds() - - heartbeat_counts = {rid: len(hbs) for rid, hbs in heartbeats_by_runtime.items()} - lifespans = {rid: lifespan(hbs) for rid, hbs in heartbeats_by_runtime.items()} - longest = max(lifespans.values(), default=0.0) - measurable_runtimes = { - rid: hbs for rid, hbs in heartbeats_by_runtime.items() if len(hbs) > 2 and lifespans[rid] >= longest * 0.5 - } - assert measurable_runtimes, ( + delays_by_runtime, heartbeat_counts = heartbeat_delays_by_runtime(interfaces.library.get_telemetry_data()) + assert delays_by_runtime, ( f"No runtime emitted enough heartbeats to check delays (runtimes seen: {heartbeat_counts})" ) - for runtime_id, heartbeats in measurable_runtimes.items(): - logger.debug(f"Heartbeats for runtime {runtime_id}:") - - # In theory, it's sorted. Let be safe - heartbeats.sort(key=lambda data: isoparse(data["request"]["timestamp_start"])) - - prev_message_time = None - delays: list[float] = [] - for data in heartbeats: - curr_message_time = isoparse(data["request"]["timestamp_start"]) - if prev_message_time is None: - logger.debug(f" * {data['log_filename']}: {curr_message_time}") - else: - delay = (curr_message_time - prev_message_time).total_seconds() - logger.debug(f" * {data['log_filename']}: {curr_message_time} => {delay}s ellapsed") - delays.append(delay) - - prev_message_time = curr_message_time - - delays_by_runtime[runtime_id] = delays + for runtime_id, delays in delays_by_runtime.items(): + logger.debug(f"Heartbeat delays for runtime {runtime_id}: {delays}") return delays_by_runtime diff --git a/tests/test_the_test/test_telemetry_heartbeat.py b/tests/test_the_test/test_telemetry_heartbeat.py new file mode 100644 index 00000000000..c9c79805ad9 --- /dev/null +++ b/tests/test_the_test/test_telemetry_heartbeat.py @@ -0,0 +1,56 @@ +from datetime import datetime, timedelta, UTC +from typing import Any + +import pytest + +from tests._telemetry_heartbeat import heartbeat_delays_by_runtime + + +pytestmark = pytest.mark.scenario("TEST_THE_TEST") + +BASE_TIME = datetime(2026, 1, 1, tzinfo=UTC) + + +def _heartbeat(runtime_id: str, seq_id: int, offset: float) -> dict[str, Any]: + timestamp = BASE_TIME + timedelta(seconds=offset) + return { + "request": { + "timestamp_start": timestamp.isoformat(), + "content": { + "request_type": "app-heartbeat", + "runtime_id": runtime_id, + "seq_id": seq_id, + }, + }, + "log_filename": f"{runtime_id}-{seq_id}-{offset}.json", + } + + +def test_fork_duplicate_does_not_affect_parent_cadence() -> None: + messages = [ + _heartbeat("parent", 1, 0), + _heartbeat("parent", 2, 2), + _heartbeat("parent", 2, 2.05), + _heartbeat("parent", 3, 4), + _heartbeat("parent", 4, 6), + _heartbeat("child", 1, 2.1), + _heartbeat("child", 2, 4.1), + ] + delays, heartbeat_counts = heartbeat_delays_by_runtime(messages) + + assert heartbeat_counts == {"parent": 4, "child": 2} + assert set(delays) == {"parent"} + assert delays["parent"] == pytest.approx([2.0, 2.0, 2.0]) + + +def test_distinct_fast_heartbeats_remain_measurable() -> None: + messages = [ + _heartbeat("parent", 1, 0), + _heartbeat("parent", 2, 1), + _heartbeat("parent", 3, 2), + _heartbeat("parent", 4, 3), + ] + delays, heartbeat_counts = heartbeat_delays_by_runtime(messages) + + assert heartbeat_counts == {"parent": 4} + assert delays["parent"] == pytest.approx([1.0, 1.0, 1.0]) From cfb4e71f881372234d06be5785bd5ac62bb7c7cc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pawe=C5=82=20Chojnacki?= Date: Mon, 13 Jul 2026 23:04:07 +0200 Subject: [PATCH 2/7] test(telemetry): follow test file naming convention --- tests/test_telemetry.py | 2 +- ...telemetry_heartbeat.py => test_telemetry_heartbeat_utils.py} | 0 tests/test_the_test/test_telemetry_heartbeat.py | 2 +- 3 files changed, 2 insertions(+), 2 deletions(-) rename tests/{_telemetry_heartbeat.py => test_telemetry_heartbeat_utils.py} (100%) diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index 488446c445e..456600411c8 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -9,7 +9,7 @@ from dateutil.parser import isoparse -from tests._telemetry_heartbeat import heartbeat_delays_by_runtime +from tests.test_telemetry_heartbeat_utils import heartbeat_delays_by_runtime from utils import bug, context, features, interfaces, logger, rfc, scenarios, weblog from utils.interfaces._misc_validators import HeadersMatchValidator, HeadersPresenceValidator from utils.telemetry import get_lang_configs, load_telemetry_json diff --git a/tests/_telemetry_heartbeat.py b/tests/test_telemetry_heartbeat_utils.py similarity index 100% rename from tests/_telemetry_heartbeat.py rename to tests/test_telemetry_heartbeat_utils.py diff --git a/tests/test_the_test/test_telemetry_heartbeat.py b/tests/test_the_test/test_telemetry_heartbeat.py index c9c79805ad9..53910e9df0c 100644 --- a/tests/test_the_test/test_telemetry_heartbeat.py +++ b/tests/test_the_test/test_telemetry_heartbeat.py @@ -3,7 +3,7 @@ import pytest -from tests._telemetry_heartbeat import heartbeat_delays_by_runtime +from tests.test_telemetry_heartbeat_utils import heartbeat_delays_by_runtime pytestmark = pytest.mark.scenario("TEST_THE_TEST") From 94f46fde228829c9b28e3e7e6a085ed8165f5cf0 Mon Sep 17 00:00:00 2001 From: Pawel Chojnacki Date: Thu, 16 Jul 2026 16:03:03 +0200 Subject: [PATCH 3/7] test(telemetry): unskip heartbeat delay on uwsgi --- manifests/python.yml | 3 --- 1 file changed, 3 deletions(-) diff --git a/manifests/python.yml b/manifests/python.yml index 354b2993ec8..dae824bf09e 100644 --- a/manifests/python.yml +++ b/manifests/python.yml @@ -2510,9 +2510,6 @@ manifest: tests/test_telemetry.py::Test_Telemetry: v1.16.0 tests/test_telemetry.py::Test_Telemetry::test_api_still_v1: irrelevant tests/test_telemetry.py::Test_Telemetry::test_app_dependencies_loaded: irrelevant - tests/test_telemetry.py::Test_Telemetry::test_app_heartbeats_delays: - - weblog_declaration: - uwsgi-poc: flaky (APMAPI-2167) tests/test_telemetry.py::Test_Telemetry::test_app_product_change: missing_feature (Weblog GET/enable_product and app-product-change event is not implemented yet.) tests/test_telemetry.py::Test_Telemetry::test_app_started_is_first_message: - declaration: flaky (APMAPI-1858) From 34b0516bc594faf6d8a3c7b74a246ca924a42657 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Thu, 16 Jul 2026 10:16:53 -0400 Subject: [PATCH 4/7] test(telemetry): guard heartbeat dedup against message-batch collisions Dedup on (runtime_id, seq_id) alone treats a flattened message-batch entry as if it were a standalone message, but flattening keeps the outer batch's seq_id on every entry it contains. Two distinct heartbeats sent in the same batch would therefore collide and one would be silently dropped. Switch to unflattened telemetry data and key the dedup on (seq_id, batch_index) instead, and document why a seq_id collision is a reliable signal for a fork clone in this codebase (/spawn_child does a raw os.fork(), so parent and child share a runtime_id while each keeps its own seq_id counter copied from the same value at fork time). Co-authored-by: Cursor --- tests/test_telemetry.py | 4 +- tests/test_telemetry_heartbeat_utils.py | 46 ++++++++++++++----- .../test_the_test/test_telemetry_heartbeat.py | 35 ++++++++++++++ 3 files changed, 72 insertions(+), 13 deletions(-) diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index 456600411c8..7a2c7bd466f 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -319,7 +319,9 @@ def _get_heartbeat_delays_by_runtime() -> dict[str, list[float]]: The value is a list of delay observed on this runtime id """ - delays_by_runtime, heartbeat_counts = heartbeat_delays_by_runtime(interfaces.library.get_telemetry_data()) + delays_by_runtime, heartbeat_counts = heartbeat_delays_by_runtime( + interfaces.library.get_telemetry_data(flatten_message_batches=False) + ) assert delays_by_runtime, ( f"No runtime emitted enough heartbeats to check delays (runtimes seen: {heartbeat_counts})" ) diff --git a/tests/test_telemetry_heartbeat_utils.py b/tests/test_telemetry_heartbeat_utils.py index 7a98414671f..98f012453fe 100644 --- a/tests/test_telemetry_heartbeat_utils.py +++ b/tests/test_telemetry_heartbeat_utils.py @@ -9,30 +9,52 @@ def heartbeat_delays_by_runtime( telemetry_data: Iterable[dict[str, Any]], ) -> tuple[dict[str, list[float]], dict[str, int]]: - """Return heartbeat delays and logical heartbeat counts grouped by runtime ID.""" - heartbeats_by_runtime: dict[str, dict[int, dict[str, Any]]] = defaultdict(dict) + """Return heartbeat delays and logical heartbeat counts grouped by runtime ID. + + Expects unflattened telemetry data (`get_telemetry_data(flatten_message_batches=False)`) + so that heartbeats packed inside a message-batch keep their position in the batch: a + flattened message-batch entry inherits the outer batch's seq_id, so two distinct + heartbeats sent in the same batch would otherwise collide on (runtime_id, seq_id) alone. + """ + heartbeats_by_runtime: dict[str, dict[tuple[int, int], dict[str, Any]]] = defaultdict(dict) for data in telemetry_data: content: dict[str, Any] = data["request"]["content"] - if content.get("request_type") != "app-heartbeat": - continue + runtime_id: str = content.get("runtime_id", "") + seq_id: int = content.get("seq_id", 0) + + if content.get("request_type") == "message-batch": + entries = list(enumerate(content.get("payload", []))) + else: + entries = [(0, content)] - runtime_id: str = content["runtime_id"] - seq_id: int = content["seq_id"] + for batch_index, entry in entries: + if entry.get("request_type") != "app-heartbeat": + continue - # A retry or a fork clone can send the same logical heartbeat more than once. - # It must contribute only once to the runtime's observed cadence. - heartbeats_by_runtime[runtime_id].setdefault(seq_id, data) + # A retry resends the same message unchanged (same seq_id, same position in its + # batch, if any) and must contribute only once to the runtime's observed cadence. + # + # A forked child briefly shares its parent's runtime_id (until it regenerates its + # own, per the Stable Service Instance Identifier RFC) but keeps its own seq_id + # counter, copied from the parent's in-memory state at fork time. Since that + # counter isn't coordinated across the fork boundary, parent and child can + # independently emit a heartbeat with the same seq_id from that shared starting + # value. Deduping on (seq_id, batch_index) catches this the same way it catches a + # real retry -- it's a coincidence of the copied counter state, not a guaranteed + # invariant, but it's what lets us tell "fork clone" apart from "genuinely fast + # heartbeat" without tracking PIDs. + heartbeats_by_runtime[runtime_id].setdefault((seq_id, batch_index), data) heartbeat_counts = {runtime_id: len(heartbeats) for runtime_id, heartbeats in heartbeats_by_runtime.items()} delays_by_runtime: dict[str, list[float]] = {} - for runtime_id, heartbeats_by_seq_id in heartbeats_by_runtime.items(): - if len(heartbeats_by_seq_id) <= 2: + for runtime_id, heartbeats_by_key in heartbeats_by_runtime.items(): + if len(heartbeats_by_key) <= 2: continue heartbeats = sorted( - heartbeats_by_seq_id.values(), + heartbeats_by_key.values(), key=lambda data: isoparse(data["request"]["timestamp_start"]), ) times = [isoparse(data["request"]["timestamp_start"]) for data in heartbeats] diff --git a/tests/test_the_test/test_telemetry_heartbeat.py b/tests/test_the_test/test_telemetry_heartbeat.py index 53910e9df0c..96a207ab217 100644 --- a/tests/test_the_test/test_telemetry_heartbeat.py +++ b/tests/test_the_test/test_telemetry_heartbeat.py @@ -26,6 +26,22 @@ def _heartbeat(runtime_id: str, seq_id: int, offset: float) -> dict[str, Any]: } +def _message_batch(runtime_id: str, seq_id: int, offset: float, heartbeat_count: int) -> dict[str, Any]: + timestamp = BASE_TIME + timedelta(seconds=offset) + return { + "request": { + "timestamp_start": timestamp.isoformat(), + "content": { + "request_type": "message-batch", + "runtime_id": runtime_id, + "seq_id": seq_id, + "payload": [{"request_type": "app-heartbeat", "payload": {}} for _ in range(heartbeat_count)], + }, + }, + "log_filename": f"{runtime_id}-{seq_id}-{offset}-batch.json", + } + + def test_fork_duplicate_does_not_affect_parent_cadence() -> None: messages = [ _heartbeat("parent", 1, 0), @@ -54,3 +70,22 @@ def test_distinct_fast_heartbeats_remain_measurable() -> None: assert heartbeat_counts == {"parent": 4} assert delays["parent"] == pytest.approx([1.0, 1.0, 1.0]) + + +def test_message_batch_heartbeats_are_not_collapsed() -> None: + """Two distinct heartbeats packed into the same message-batch keep a distinct identity. + + A flattened message-batch entry inherits the outer batch's seq_id, so deduping on + (runtime_id, seq_id) alone would silently drop one of them. Deduping on + (seq_id, batch_index) instead keeps both. + """ + messages = [ + _heartbeat("solo", 1, 0), + _message_batch("solo", 2, 2, heartbeat_count=2), + _heartbeat("solo", 3, 4), + _heartbeat("solo", 4, 6), + ] + delays, heartbeat_counts = heartbeat_delays_by_runtime(messages) + + assert heartbeat_counts == {"solo": 5} + assert delays["solo"] == pytest.approx([2.0, 0.0, 2.0, 2.0]) From a14b892a8359e9c53b3a491d2ad5fa2aeb14006d Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Thu, 16 Jul 2026 10:20:07 -0400 Subject: [PATCH 5/7] test(telemetry): remove message-batch regression test Keep the batch-index-aware dedup fix, but drop the dedicated regression test for the message-batch collision scenario. Co-authored-by: Cursor --- .../test_the_test/test_telemetry_heartbeat.py | 35 ------------------- 1 file changed, 35 deletions(-) diff --git a/tests/test_the_test/test_telemetry_heartbeat.py b/tests/test_the_test/test_telemetry_heartbeat.py index 96a207ab217..53910e9df0c 100644 --- a/tests/test_the_test/test_telemetry_heartbeat.py +++ b/tests/test_the_test/test_telemetry_heartbeat.py @@ -26,22 +26,6 @@ def _heartbeat(runtime_id: str, seq_id: int, offset: float) -> dict[str, Any]: } -def _message_batch(runtime_id: str, seq_id: int, offset: float, heartbeat_count: int) -> dict[str, Any]: - timestamp = BASE_TIME + timedelta(seconds=offset) - return { - "request": { - "timestamp_start": timestamp.isoformat(), - "content": { - "request_type": "message-batch", - "runtime_id": runtime_id, - "seq_id": seq_id, - "payload": [{"request_type": "app-heartbeat", "payload": {}} for _ in range(heartbeat_count)], - }, - }, - "log_filename": f"{runtime_id}-{seq_id}-{offset}-batch.json", - } - - def test_fork_duplicate_does_not_affect_parent_cadence() -> None: messages = [ _heartbeat("parent", 1, 0), @@ -70,22 +54,3 @@ def test_distinct_fast_heartbeats_remain_measurable() -> None: assert heartbeat_counts == {"parent": 4} assert delays["parent"] == pytest.approx([1.0, 1.0, 1.0]) - - -def test_message_batch_heartbeats_are_not_collapsed() -> None: - """Two distinct heartbeats packed into the same message-batch keep a distinct identity. - - A flattened message-batch entry inherits the outer batch's seq_id, so deduping on - (runtime_id, seq_id) alone would silently drop one of them. Deduping on - (seq_id, batch_index) instead keeps both. - """ - messages = [ - _heartbeat("solo", 1, 0), - _message_batch("solo", 2, 2, heartbeat_count=2), - _heartbeat("solo", 3, 4), - _heartbeat("solo", 4, 6), - ] - delays, heartbeat_counts = heartbeat_delays_by_runtime(messages) - - assert heartbeat_counts == {"solo": 5} - assert delays["solo"] == pytest.approx([2.0, 0.0, 2.0, 2.0]) From 1319851f5cd6e0f7f82ed12bf9de495c23256e47 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Thu, 16 Jul 2026 11:24:13 -0400 Subject: [PATCH 6/7] test(telemetry): exclude short-lived runtimes from heartbeat cadence check Deduping by (seq_id, batch_index) only catches retries and fork clones that share a seq_id. Locally reproducing the flake (10+ runs of the full Test_Telemetry class against python/uwsgi-poc) showed a different, more common failure: a forked child from the session-id tests exits ~1 heartbeat interval after starting, but can emit an extra out-of-cadence heartbeat as part of its shutdown flush -- a real, distinct message with its own seq_id, so dedup can't touch it. With only 2-3 total samples, that single anomalous gap dominates the average and fails the assertion regardless of dedup. Add a min_lifespan floor (2x the configured heartbeat interval): runtimes that didn't live long enough to give a statistically meaningful cadence sample are excluded from measurement entirely, while long-lived runtimes with the same kind of anomaly are still measured and can still fail. This reproduced the flake reliably before the change and passed 10/10 local runs after it. Co-authored-by: Cursor --- tests/test_telemetry.py | 9 +++- tests/test_telemetry_heartbeat_utils.py | 16 +++++++ .../test_the_test/test_telemetry_heartbeat.py | 43 +++++++++++++++++++ 3 files changed, 67 insertions(+), 1 deletion(-) diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index 7a2c7bd466f..02851f00c3a 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -320,7 +320,14 @@ def _get_heartbeat_delays_by_runtime() -> dict[str, list[float]]: """ delays_by_runtime, heartbeat_counts = heartbeat_delays_by_runtime( - interfaces.library.get_telemetry_data(flatten_message_batches=False) + interfaces.library.get_telemetry_data(flatten_message_batches=False), + # A runtime needs to have been alive for a couple of heartbeat intervals before + # its cadence is trustworthy: with only 2-3 samples, a single heartbeat sent near + # process shutdown (e.g. a final flush) can single-handedly skew the average. + # This is comfortably above the ~1 interval lifespan of the short-lived children + # spawned by the session-id tests, without requiring more real time to elapse + # than a normal long-lived runtime accumulates in a single test session. + min_lifespan=context.telemetry_heartbeat_interval * 2, ) assert delays_by_runtime, ( f"No runtime emitted enough heartbeats to check delays (runtimes seen: {heartbeat_counts})" diff --git a/tests/test_telemetry_heartbeat_utils.py b/tests/test_telemetry_heartbeat_utils.py index 98f012453fe..960066b3517 100644 --- a/tests/test_telemetry_heartbeat_utils.py +++ b/tests/test_telemetry_heartbeat_utils.py @@ -8,6 +8,8 @@ def heartbeat_delays_by_runtime( telemetry_data: Iterable[dict[str, Any]], + *, + min_lifespan: float = 0.0, ) -> tuple[dict[str, list[float]], dict[str, int]]: """Return heartbeat delays and logical heartbeat counts grouped by runtime ID. @@ -15,6 +17,16 @@ def heartbeat_delays_by_runtime( so that heartbeats packed inside a message-batch keep their position in the batch: a flattened message-batch entry inherits the outer batch's seq_id, so two distinct heartbeats sent in the same batch would otherwise collide on (runtime_id, seq_id) alone. + + `min_lifespan` excludes runtimes whose heartbeats span less than that many seconds + (e.g. `telemetry_heartbeat_interval * 3`). A process that lived only a fraction of the + heartbeat interval before exiting (e.g. a forked child spawned by the session-id tests) + doesn't provide a reliable cadence sample: some tracers emit an extra, out-of-cadence + heartbeat as part of their shutdown flush, and with only 2-3 total samples that single + anomalous gap dominates the average delay. This is a real, distinct message (its own + seq_id, not a retry/clone) so deduping can't catch it -- only requiring enough elapsed + time to average it out can. A long-lived runtime with the same anomaly is still measured + and can still fail: this only excludes runtimes too short-lived to measure at all. """ heartbeats_by_runtime: dict[str, dict[tuple[int, int], dict[str, Any]]] = defaultdict(dict) @@ -58,6 +70,10 @@ def heartbeat_delays_by_runtime( key=lambda data: isoparse(data["request"]["timestamp_start"]), ) times = [isoparse(data["request"]["timestamp_start"]) for data in heartbeats] + lifespan = (times[-1] - times[0]).total_seconds() + if lifespan < min_lifespan: + continue + delays_by_runtime[runtime_id] = [ (current - previous).total_seconds() for previous, current in itertools.pairwise(times) ] diff --git a/tests/test_the_test/test_telemetry_heartbeat.py b/tests/test_the_test/test_telemetry_heartbeat.py index 53910e9df0c..294e5419e95 100644 --- a/tests/test_the_test/test_telemetry_heartbeat.py +++ b/tests/test_the_test/test_telemetry_heartbeat.py @@ -54,3 +54,46 @@ def test_distinct_fast_heartbeats_remain_measurable() -> None: assert heartbeat_counts == {"parent": 4} assert delays["parent"] == pytest.approx([1.0, 1.0, 1.0]) + + +def test_short_lived_runtime_excluded_by_min_lifespan() -> None: + """A forked child that exits shortly after starting can emit a shutdown-triggered + heartbeat with a distinct seq_id (not a retry/clone, so dedup can't catch it), which + skews the average for a runtime with only a couple of samples. With too few real + heartbeat intervals elapsed, the runtime shouldn't be measured at all. + """ + messages = [ + _heartbeat("parent", 1, 0), + _heartbeat("parent", 2, 2), + _heartbeat("parent", 3, 4), + _heartbeat("parent", 4, 6), + _heartbeat("parent", 5, 8), + # child lives ~2.05s: two normal-ish heartbeats, then an out-of-cadence one + # right before it exits + _heartbeat("child", 1, 0.1), + _heartbeat("child", 2, 2.05), + _heartbeat("child", 3, 2.12), + ] + delays, heartbeat_counts = heartbeat_delays_by_runtime(messages, min_lifespan=6.0) + + assert heartbeat_counts == {"parent": 5, "child": 3} + assert set(delays) == {"parent"} + + +def test_long_lived_runtime_anomaly_still_measured() -> None: + """The min_lifespan floor only excludes runtimes too short-lived to measure at all -- + a long-lived runtime with the same kind of anomalous fast heartbeat is still measured + (and can still fail the caller's assertion), since that's a real signal worth catching. + """ + messages = [ + _heartbeat("parent", 1, 0), + _heartbeat("parent", 2, 2), + _heartbeat("parent", 3, 4), + _heartbeat("parent", 4, 6), + _heartbeat("parent", 5, 8), + _heartbeat("parent", 6, 8.05), # anomalous fast heartbeat, but lifespan is 8.05s + ] + delays, heartbeat_counts = heartbeat_delays_by_runtime(messages, min_lifespan=6.0) + + assert heartbeat_counts == {"parent": 6} + assert delays["parent"] == pytest.approx([2.0, 2.0, 2.0, 2.0, 0.05]) From cc5c177b356a153976ebfb537ba6f363017fb0dc Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Thu, 16 Jul 2026 12:16:25 -0400 Subject: [PATCH 7/7] test(telemetry): simplify heartbeat comments, fix CI scenario selection Trim the heartbeat dedup/lifespan comments down to their essential reasoning, and add a selector rule so that changes to the standalone test_telemetry_heartbeat_utils.py helper (which has no test items of its own) correctly select the DEFAULT and TEST_THE_TEST scenarios in CI's diff-based scenario selection. Co-authored-by: Cursor --- tests/test_telemetry.py | 9 ++-- tests/test_telemetry_heartbeat_utils.py | 42 +++++++------------ .../test_the_test/test_telemetry_heartbeat.py | 12 +++--- .../scripts/libraries_and_scenarios_rules.yml | 6 +++ 4 files changed, 30 insertions(+), 39 deletions(-) diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index 02851f00c3a..cbf2c482e40 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -321,12 +321,9 @@ def _get_heartbeat_delays_by_runtime() -> dict[str, list[float]]: delays_by_runtime, heartbeat_counts = heartbeat_delays_by_runtime( interfaces.library.get_telemetry_data(flatten_message_batches=False), - # A runtime needs to have been alive for a couple of heartbeat intervals before - # its cadence is trustworthy: with only 2-3 samples, a single heartbeat sent near - # process shutdown (e.g. a final flush) can single-handedly skew the average. - # This is comfortably above the ~1 interval lifespan of the short-lived children - # spawned by the session-id tests, without requiring more real time to elapse - # than a normal long-lived runtime accumulates in a single test session. + # Require 2 intervals of lifespan so a short-lived process (e.g. a forked child + # from the session-id tests) can't skew the average with just a couple of + # samples. A normal long-lived runtime clears this easily. min_lifespan=context.telemetry_heartbeat_interval * 2, ) assert delays_by_runtime, ( diff --git a/tests/test_telemetry_heartbeat_utils.py b/tests/test_telemetry_heartbeat_utils.py index 960066b3517..c6c22f20453 100644 --- a/tests/test_telemetry_heartbeat_utils.py +++ b/tests/test_telemetry_heartbeat_utils.py @@ -11,22 +11,18 @@ def heartbeat_delays_by_runtime( *, min_lifespan: float = 0.0, ) -> tuple[dict[str, list[float]], dict[str, int]]: - """Return heartbeat delays and logical heartbeat counts grouped by runtime ID. + """Return heartbeat delays and heartbeat counts, grouped by runtime ID. - Expects unflattened telemetry data (`get_telemetry_data(flatten_message_batches=False)`) - so that heartbeats packed inside a message-batch keep their position in the batch: a - flattened message-batch entry inherits the outer batch's seq_id, so two distinct - heartbeats sent in the same batch would otherwise collide on (runtime_id, seq_id) alone. + Expects unflattened telemetry data (`get_telemetry_data(flatten_message_batches=False)`): + a flattened message-batch entry inherits the outer batch's seq_id, so heartbeats packed + in the same batch would otherwise collide on (runtime_id, seq_id) alone and get deduped + into one. - `min_lifespan` excludes runtimes whose heartbeats span less than that many seconds - (e.g. `telemetry_heartbeat_interval * 3`). A process that lived only a fraction of the - heartbeat interval before exiting (e.g. a forked child spawned by the session-id tests) - doesn't provide a reliable cadence sample: some tracers emit an extra, out-of-cadence - heartbeat as part of their shutdown flush, and with only 2-3 total samples that single - anomalous gap dominates the average delay. This is a real, distinct message (its own - seq_id, not a retry/clone) so deduping can't catch it -- only requiring enough elapsed - time to average it out can. A long-lived runtime with the same anomaly is still measured - and can still fail: this only excludes runtimes too short-lived to measure at all. + `min_lifespan` drops runtimes whose heartbeats span less time than that (e.g. a forked + child that exits after only 2-3 heartbeats). With so few samples, a real (non-duplicate) + heartbeat sent during shutdown can dominate the average delay -- dropping the runtime is + the only way to avoid measuring it. A long-lived runtime with the same anomaly is still + measured, and can still fail. """ heartbeats_by_runtime: dict[str, dict[tuple[int, int], dict[str, Any]]] = defaultdict(dict) @@ -44,18 +40,12 @@ def heartbeat_delays_by_runtime( if entry.get("request_type") != "app-heartbeat": continue - # A retry resends the same message unchanged (same seq_id, same position in its - # batch, if any) and must contribute only once to the runtime's observed cadence. - # - # A forked child briefly shares its parent's runtime_id (until it regenerates its - # own, per the Stable Service Instance Identifier RFC) but keeps its own seq_id - # counter, copied from the parent's in-memory state at fork time. Since that - # counter isn't coordinated across the fork boundary, parent and child can - # independently emit a heartbeat with the same seq_id from that shared starting - # value. Deduping on (seq_id, batch_index) catches this the same way it catches a - # real retry -- it's a coincidence of the copied counter state, not a guaranteed - # invariant, but it's what lets us tell "fork clone" apart from "genuinely fast - # heartbeat" without tracking PIDs. + # Dedup key (seq_id, batch_index) catches two cases: a real retry (identical + # resend), and a forked child's clone heartbeat. A child briefly shares its + # parent's runtime_id and copies its seq_id counter at fork time, so it can + # independently emit a heartbeat with the same seq_id. Not a guaranteed + # invariant, but the best signal we have to tell "fork clone" apart from + # "genuinely fast heartbeat" without tracking PIDs. heartbeats_by_runtime[runtime_id].setdefault((seq_id, batch_index), data) heartbeat_counts = {runtime_id: len(heartbeats) for runtime_id, heartbeats in heartbeats_by_runtime.items()} diff --git a/tests/test_the_test/test_telemetry_heartbeat.py b/tests/test_the_test/test_telemetry_heartbeat.py index 294e5419e95..d5f1e1b8dcc 100644 --- a/tests/test_the_test/test_telemetry_heartbeat.py +++ b/tests/test_the_test/test_telemetry_heartbeat.py @@ -57,10 +57,9 @@ def test_distinct_fast_heartbeats_remain_measurable() -> None: def test_short_lived_runtime_excluded_by_min_lifespan() -> None: - """A forked child that exits shortly after starting can emit a shutdown-triggered - heartbeat with a distinct seq_id (not a retry/clone, so dedup can't catch it), which - skews the average for a runtime with only a couple of samples. With too few real - heartbeat intervals elapsed, the runtime shouldn't be measured at all. + """A short-lived runtime with too few samples shouldn't be measured at all: a shutdown + heartbeat right before exit isn't a duplicate, so dedup can't drop it, and with only + 2-3 samples it would skew the average. """ messages = [ _heartbeat("parent", 1, 0), @@ -81,9 +80,8 @@ def test_short_lived_runtime_excluded_by_min_lifespan() -> None: def test_long_lived_runtime_anomaly_still_measured() -> None: - """The min_lifespan floor only excludes runtimes too short-lived to measure at all -- - a long-lived runtime with the same kind of anomalous fast heartbeat is still measured - (and can still fail the caller's assertion), since that's a real signal worth catching. + """min_lifespan only excludes runtimes too short-lived to measure. A long-lived runtime + with the same anomalous fast heartbeat is still measured, and can still fail. """ messages = [ _heartbeat("parent", 1, 0), diff --git a/utils/scripts/libraries_and_scenarios_rules.yml b/utils/scripts/libraries_and_scenarios_rules.yml index 11ed0361615..44ebdc207cd 100644 --- a/utils/scripts/libraries_and_scenarios_rules.yml +++ b/utils/scripts/libraries_and_scenarios_rules.yml @@ -547,6 +547,12 @@ patterns: scenario_groups: null libraries: null + # This helper has no test items of its own, so it's invisible to the scenario map that + # drives dynamic selection below. List its consumers' scenarios explicitly. + tests/test_telemetry_heartbeat_utils.py: + scenario_groups: null + scenarios: [DEFAULT, TEST_THE_TEST] + tests/fuzzer/*: scenario_groups: exotics libraries: null