test(ffe): repair context bounds and isolate overload accounting

.github/workflows/run-end-to-end.yml

@@ -311,6 +311,9 @@ jobs:
     - name: Run FEATURE_FLAGGING_AND_EXPERIMENTATION_AGENTLESS scenario
       if: steps.build.outcome == 'success' && !cancelled() && contains(inputs.scenarios, '"FEATURE_FLAGGING_AND_EXPERIMENTATION_AGENTLESS"')
       run: ./run.sh FEATURE_FLAGGING_AND_EXPERIMENTATION_AGENTLESS
+    - name: Run FEATURE_FLAGGING_EVP_STRESS scenario
+      if: steps.build.outcome == 'success' && !cancelled() && contains(inputs.scenarios, '"FEATURE_FLAGGING_EVP_STRESS"')
+      run: ./run.sh FEATURE_FLAGGING_EVP_STRESS
     - name: Run FEATURE_FLAGGING_AND_EXPERIMENTATION_AGENTLESS_DIRECT scenario
       if: steps.build.outcome == 'success' && !cancelled() && contains(inputs.scenarios, '"FEATURE_FLAGGING_AND_EXPERIMENTATION_AGENTLESS_DIRECT"')
       run: ./run.sh FEATURE_FLAGGING_AND_EXPERIMENTATION_AGENTLESS_DIRECT

docs/understand/weblogs/end-to-end_weblog.md

@@ -1153,6 +1153,13 @@ accept a JSON body with these fields:
 - `targetingKey`: the evaluation subject key.
 - `attributes`: flat scalar targeting attributes.
 
+For the high-cardinality FFE workload, the endpoint also accepts a non-empty
+`targetingKeys` array. Evaluate the same flag once per key through the public SDK
+API, using the same attributes and default value. The response includes `count`
+(the number of targeting keys in the request) and the final evaluation's `value`.
+Do not replace the loop with a direct telemetry submission or change SDK flush
+timing. When the array is absent or empty, evaluate the single `targetingKey`.
+
 The response must be JSON and include at least `value` and `reason`. Error
 responses should also include `errorCode` and `errorMessage`.
 

manifests/golang.yml

@@ -1768,6 +1768,7 @@ manifest:
   tests/ffe/test_flag_eval_evp.py: v2.10.0-dev
   tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_Egress_Agentless_Direct: v2.12.0-dev
   tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_Egress_Agentless_Sidecar: v2.12.0-dev
+  tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_High_Cardinality_Accounting: v2.11.0
   tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_ObserveFullData_Absent_Hashed: v2.11.0-dev
   tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_ObserveFullData_False_Hashed: v2.11.0-dev
   tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_ObserveFullData_True_Unhashed: v2.11.0-dev  # Easy win for chi, echo, gin, net-http, net-http-orchestrion, uds-echo and version 2.10.0

manifests/php.yml

@@ -1285,7 +1285,6 @@ manifest:
       excluded_weblog: [apache-mod-8.0-zts, apache-mod-7.1-zts, apache-mod-8.0, apache-mod-7.4, php-fpm-7.2, php-fpm-8.1, apache-mod-7.0, apache-mod-7.0-zts, php-fpm-7.0, php-fpm-8.0, php-fpm-8.2, apache-mod-7.2-zts, apache-mod-8.1-zts, apache-mod-7.1, php-fpm-8.5, apache-mod-8.2-zts, php-fpm-7.3, php-fpm-7.4, apache-mod-7.2, apache-mod-7.4-zts, apache-mod-8.2, apache-mod-7.3, apache-mod-8.1, php-fpm-7.1, apache-mod-7.3-zts]
   tests/ffe/test_flag_eval_evp.py: missing_feature (FFL-2446)
   tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_Burst_Aggregation: bug (FFL-2676)
-  tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_Degradation: bug (FFL-2676)
   tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_High_Cardinality_Aggregation: bug (FFL-2676)
   tests/ffe/test_flag_eval_metrics.py:
     - weblog_declaration:

manifests/python.yml

@@ -2306,10 +2306,6 @@ manifest:
     - declaration: missing_feature (FFL-2446)
     - weblog_declaration:
         aiohttp: missing_feature (new weblog, tests are enabled with easy wins)
-  tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_Degradation:
-    - declaration: missing_feature (FFL-2446)
-    - weblog_declaration:
-        aiohttp: missing_feature (new weblog, tests are enabled with easy wins)
   tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_Egress_Agentless_Direct:
     - weblog_declaration:
         "*": missing_feature (Agentless EVP enablement is validated only on flask-poc)
@@ -2324,6 +2320,10 @@ manifest:
       weblog: [django-poc, uwsgi-poc, uds-flask, python3.12, django-py3.13, aiohttp, flask-poc, fastapi, tornado]
     - declaration: missing_feature (FFL-2446)
       excluded_weblog: [django-poc, uwsgi-poc, uds-flask, python3.12, django-py3.13, aiohttp, flask-poc, fastapi, tornado]
+  tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_High_Cardinality_Accounting:
+    - declaration: missing_feature (FFL-2446)
+    - weblog_declaration:
+        aiohttp: missing_feature (new weblog, tests are enabled with easy wins)
   tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_High_Cardinality_Aggregation:
     - declaration: missing_feature (FFL-2446)
     - weblog_declaration:

tests/ffe/README.md

@@ -143,6 +143,28 @@ environment installed, run them explicitly:
 ```
 Framework mock-backend and lifecycle tests stay in `tests/test_the_test/`.
 
+## Bounds and overload coverage
+
+With full-data consent, the context tests require all 256 valid attributes to
+survive unchanged at the field-count limit. Over-limit contexts may retain an
+unchanged subset of at most 256 attributes or be omitted entirely. A separate
+string-length test requires oversized values to be omitted while valid fields survive.
+
+Run the high-cardinality accounting test in its dedicated scenario:
+
+```sh
+./run.sh FEATURE_FLAGGING_EVP_STRESS --library <language>
+```
+
+Keep this scenario limited to one evaluation workload: its drop counters are
+process-wide. Delivered evaluation counts (including degraded delivery) plus
+accepted reported drops must equal attempted evaluations, and some delivery is
+required. See the [accounting helper](utils/evp.py) for accepted counters and validation.
+Periodic flushes may prevent degradation, so the test validates degraded rows when
+present but does not require them. Deterministic degradation-threshold checks belong
+in SDK-local tests. Drop counters must report lost evaluations, including those
+represented by aggregated rows.
+
 ---
 
 # Eval Metrics Implementation Guide

tests/ffe/test_flag_eval_evp.py

@@ -7,6 +7,7 @@
 
 
 from tests.ffe.utils.evp import (
+    assert_flagevaluation_accounting,
     assert_agentless_evp_intake_request,
     assert_agentless_evp_topology,
     assert_evp_sdk_identity,
@@ -29,8 +30,7 @@
 RC_PRODUCT = "FFE_FLAGS"
 RC_PATH = f"datadog/2/{RC_PRODUCT}"
 EVP_FLAGEVALUATIONS_PATH = "/api/v2/flagevaluation"
-EVP_FULL_TIER_PER_FLAG_CAP = 10_000
-EVP_DEGRADATION_OVERFLOW_EVALS = 2_000
+EVP_HIGH_CARDINALITY_EVALS = 12_000
 
 # Fixed input/output vector for the PII-protection tests. Every SDK's unit tests
 # should assert against the same input to prove byte-identical hashing across SDKs.
@@ -420,43 +420,96 @@ def test_ffe_evp_flagevaluation_count(self) -> None:
 class Test_FFE_EVP_Flagevaluation_Context_Bounds:
     """Test that EVP evaluation context is bounded before it reaches payloads."""
 
+    def setup_ffe_evp_flagevaluation_context_within_bounds(self) -> None:
+        self.within_flag_key = "evp-context-within-bounds-flag"
+        rc.tracer_rc_state.reset().set_config(
+            f"{RC_PATH}/ffe-evp-context-within-bounds/config",
+            make_ufc_fixture(self.within_flag_key, observe_full_evaluation_data=True),
+        ).apply()
+        self.within_attributes: JSON = {f"field_{index:03d}": f"value-{index}" for index in range(256)}
+        self.within_response = evaluate_flag(
+            self.within_flag_key, targeting_key="evp-context-user", attributes=self.within_attributes
+        )
+
+    def test_ffe_evp_flagevaluation_context_within_bounds(self) -> None:
+        events = self._assert_context_evaluation(self.within_response, self.within_flag_key)
+        for _, event in events:
+            context = event.get("context")
+            assert isinstance(context, dict), "in-limit consented evaluation must retain context"
+            assert context.get("evaluation") == self.within_attributes, "in-limit context fields must survive unchanged"
+
     def setup_ffe_evp_flagevaluation_context_bounds(self) -> None:
         config_id = "ffe-evp-context-bounds"
         self.flag_key = "evp-context-bounds-flag"
-        self.oversized_field = "field_010_oversized"
-        rc.tracer_rc_state.reset().set_config(f"{RC_PATH}/{config_id}/config", make_ufc_fixture(self.flag_key)).apply()
-
-        attributes: JSON = {f"field_{index:03d}": f"value-{index}" for index in range(300)}
-        attributes[self.oversized_field] = "x" * 300
+        rc.tracer_rc_state.reset().set_config(
+            f"{RC_PATH}/{config_id}/config", make_ufc_fixture(self.flag_key, observe_full_evaluation_data=True)
+        ).apply()
 
-        self.r = evaluate_flag(self.flag_key, targeting_key="evp-context-user", attributes=attributes)
+        self.attributes: JSON = {f"field_{index:03d}": f"value-{index}" for index in range(300)}
+        self.r = evaluate_flag(self.flag_key, targeting_key="evp-context-user", attributes=self.attributes)
 
     def test_ffe_evp_flagevaluation_context_bounds(self) -> None:
-        assert self.r.status_code == 200, f"Flag evaluation failed: {self.r.text}"
-
-        events = find_evp_flagevaluation_events(self.flag_key)
-        assert events, f"Expected EVP flagevaluation event for flag {self.flag_key}"
-
-        full_context_events = 0
+        events = self._assert_context_evaluation(self.r, self.flag_key)
         for _, event in events:
-            assert_event_contract(event, self.flag_key)
-            context = event.get("context")
-            if context is None:
+            # SDKs may omit an over-limit map entirely to bound capture work.
+            # The separate in-limit case prevents unconditional context loss from passing.
+            if "context" not in event:
                 continue
-
+            context = event["context"]
             assert isinstance(context, dict), "context must be an object when present"
-            evaluation_context = context.get("evaluation")
-            if evaluation_context is None:
+            if "evaluation" not in context:
                 continue
-
-            full_context_events += 1
+            evaluation_context = context["evaluation"]
             assert isinstance(evaluation_context, dict), "context.evaluation must be an object"
             assert len(evaluation_context) <= 256, f"context.evaluation has too many fields: {len(evaluation_context)}"
-            assert self.oversized_field not in evaluation_context, "oversized string context field must be omitted"
+            # SDK maps need not preserve input order or choose the same retained subset.
+            for key, value in evaluation_context.items():
+                assert key in self.attributes, f"unexpected context field: {key}"
+                assert value == self.attributes[key], f"context field {key} changed during pruning"
+
+    @staticmethod
+    def _assert_context_evaluation(
+        response: HttpResponse, flag_key: str, targeting_key: str = "evp-context-user"
+    ) -> list[tuple[JSON, JSON]]:
+        assert response.status_code == 200, f"Flag evaluation failed: {response.text}"
+        assert json.loads(response.text).get("value") == "on-value", "context limits must not change the flag result"
+        events = find_evp_flagevaluation_events(flag_key)
+        assert_total_evaluation_count(events, 1, flag_key)
+        for _, event in events:
+            assert_event_contract(event, flag_key)
+            assert event.get("targeting_key") == targeting_key, "context limits must preserve consented identity"
+            assert object_key(event.get("variant"), "variant") == "on"
+            assert object_key(event.get("allocation"), "allocation") == "default-allocation"
+            assert "error" not in event
+            assert "runtime_default_used" not in event
+        return events
 
-        if full_context_events == 0:
-            for _, event in events:
-                assert "context" not in event, f"degraded event should omit context: {event}"
+    def setup_ffe_evp_flagevaluation_context_value_bounds(self) -> None:
+        self.value_flag_key = "evp-context-value-bounds-flag"
+        self.oversized_field = "oversized"
+        rc.tracer_rc_state.reset().set_config(
+            f"{RC_PATH}/ffe-evp-context-value-bounds/config",
+            make_ufc_fixture(self.value_flag_key, observe_full_evaluation_data=True),
+        ).apply()
+
+        # Stay below the field-count limit so it cannot hide a missing value-length check.
+        self.value_response = evaluate_flag(
+            self.value_flag_key,
+            targeting_key="evp-context-value-user",
+            attributes={"valid": "retained", self.oversized_field: "x" * 300},
+        )
+
+    def test_ffe_evp_flagevaluation_context_value_bounds(self) -> None:
+        events = self._assert_context_evaluation(
+            self.value_response, self.value_flag_key, targeting_key="evp-context-value-user"
+        )
+        for _, event in events:
+            context = event.get("context")
+            assert isinstance(context, dict), "consented evaluation must retain context"
+            evaluation_context = context.get("evaluation")
+            assert isinstance(evaluation_context, dict), "context.evaluation must be an object"
+            assert evaluation_context.get("valid") == "retained", "valid context fields must survive pruning"
+            assert self.oversized_field not in evaluation_context, "oversized string context field must be omitted"
 
 
 @scenarios.feature_flagging_and_experimentation
@@ -602,18 +655,16 @@ def test_ffe_evp_flagevaluation_high_cardinality_aggregation(self) -> None:
         assert_total_evaluation_count(events, self.eval_count, self.flag_key)
 
 
-@scenarios.feature_flagging_and_experimentation
+@scenarios.feature_flagging_evp_stress
 @features.feature_flags_evp_flagevaluation
 @scenario_crash
-class Test_FFE_EVP_Flagevaluation_Degradation:
-    """Test degraded EVP shape after the production per-flag full-tier cap is exceeded."""
-
-    def setup_ffe_evp_flagevaluation_degradation(self) -> None:
-        config_id = "ffe-evp-degradation"
-        self.flag_key = "evp-degradation-flag"
-        # The cap resets on each periodic flush. Cover two windows so a flush during
-        # the request still leaves at least one window over the cap.
-        self.eval_count = 2 * (EVP_FULL_TIER_PER_FLAG_CAP + EVP_DEGRADATION_OVERFLOW_EVALS)
+class Test_FFE_EVP_Flagevaluation_High_Cardinality_Accounting:
+    """Account for an isolated workload without assuming a single aggregation window."""
+
+    def setup_ffe_evp_flagevaluation_high_cardinality_accounting(self) -> None:
+        config_id = "ffe-evp-high-cardinality-accounting"
+        self.flag_key = "evp-high-cardinality-accounting-flag"
+        self.eval_count = EVP_HIGH_CARDINALITY_EVALS
         rc.tracer_rc_state.reset().set_config(f"{RC_PATH}/{config_id}/config", make_ufc_fixture(self.flag_key)).apply()
 
         # Keep unique keys short so the batch fits weblog request body limits.
@@ -627,24 +678,42 @@ def setup_ffe_evp_flagevaluation_degradation(self) -> None:
             )
         ]
 
-    def test_ffe_evp_flagevaluation_degradation(self) -> None:
+    def test_ffe_evp_flagevaluation_high_cardinality_accounting(self) -> None:
         for index, response in enumerate(self.responses):
             assert response.status_code == 200, f"Request {index + 1} failed: {response.text}"
+            result = json.loads(response.text)
+            assert result.get("value") == "on-value", "overload must not change the flag result"
+            assert result.get("count") == self.eval_count, "weblog must accept the whole targetingKeys batch"
 
         events = find_evp_flagevaluation_events(self.flag_key)
         assert events, f"Expected EVP flagevaluation events for flag {self.flag_key}"
 
-        degraded_events = [event for _, event in events if "context" not in event and "targeting_key" not in event]
-        assert degraded_events, f"Expected degraded EVP flagevaluation buckets for flag {self.flag_key}"
         assert_no_duplicate_visible_events(events)
-        assert_total_evaluation_count(events, self.eval_count, self.flag_key)
+        # These counters are process-wide, not flag-scoped: this class must remain
+        # the only evaluation workload in FEATURE_FLAGGING_EVP_STRESS.
+        # Tracers report rows.dropped; libdatadog's sidecar reports evaluations.dropped.
+        # Both count lost evaluations; current emitters do not report the same loss under both names.
+        drop_series = [
+            series
+            for namespace in ("tracers", "general")
+            for metric in ("flagevaluation.rows.dropped", "flagevaluation.evaluations.dropped")
+            for series in interfaces.library.get_telemetry_metric_series(namespace, metric)
+        ]
+        assert_flagevaluation_accounting(sum_evaluation_count(events), self.eval_count, drop_series)
 
-        for event in degraded_events:
+        # A run may span multiple flushes and never degrade.
+        # Every delivered row must still have a valid tier shape.
+        for _, event in events:
             assert_event_contract(event, self.flag_key)
-            assert "context" not in event, f"degraded event must omit context: {event}"
-            assert "targeting_key" not in event, f"degraded event must omit targeting_key: {event}"
+            if "targeting_key" not in event:
+                assert "context" not in event, "degraded events must omit all event context"
+            else:
+                assert isinstance(event["targeting_key"], str), "full-tier targeting_key must be a string"
+                assert event["targeting_key"], "full-tier targeting_key must be non-empty"
             assert object_key(event.get("variant"), "variant") == "on"
             assert object_key(event.get("allocation"), "allocation") == "default-allocation"
+            assert "error" not in event
+            assert "runtime_default_used" not in event
 
 
 @scenarios.feature_flagging_and_experimentation

tests/ffe/utils/evp.py

@@ -3,6 +3,7 @@
 from collections.abc import Iterable
 from dataclasses import dataclass
 from datetime import datetime
+from math import isfinite
 from pathlib import Path
 from typing import Any, Literal
 
@@ -52,6 +53,39 @@
 EVPRoute = Literal["agent", "sidecar", "direct"]
 
 
+def assert_flagevaluation_accounting(delivered: int, attempted: int, drop_series: Iterable[dict[str, Any]]) -> None:
+    """Account for isolated stress traffic; never count degradation as lost evaluations."""
+    dropped = 0
+    capacity_reasons = {
+        "reason:pre_queue_overflow",
+        "reason:queue_overflow",
+        "reason:degraded_cap",
+        "reason:payload_limit",
+    }
+    for series in drop_series:
+        assert series.get("type") == "count", "drop telemetry must be a delta count"
+        reasons = {tag for tag in series.get("tags", []) if tag.startswith("reason:")}
+        assert len(reasons) == 1, f"expected one drop reason: {reasons}"
+        assert reasons <= capacity_reasons, f"unexpected drop reason: {reasons}"
+        points = series.get("points")
+        assert isinstance(points, list), "drop metric points must be a list"
+        assert points, "drop metric must contain points"
+        for point in points:
+            assert isinstance(point, list), "invalid drop metric point"
+            assert len(point) == 2, "invalid drop metric point"
+            count = point[1]
+            assert type(count) in (int, float), "invalid drop count type"
+            assert isfinite(count), "drop count must be finite"
+            assert count >= 0, "drop count must be nonnegative"
+            assert count == int(count), "drop count must be integral"
+            dropped += int(count)
+    assert delivered > 0, "stress workload must deliver evaluations, not only report drops"
+    assert delivered + dropped == attempted, (
+        f"Expected delivered + reported drops == {attempted}, got {delivered} + {dropped}. "
+        "Missing telemetry is not evidence of an accounted drop."
+    )
+
+
 def _is_expected_pid1_executable(library_name: str, executable: str, expected: set[str]) -> bool:
     """Accept ordinary language launchers without accepting wrappers by prefix."""
     if executable in expected: