Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions docs/internal/feature-flags/rust-service-overview.md
Original file line number Diff line number Diff line change
Expand Up @@ -237,6 +237,10 @@ All values come from environment variables via the `envconfig` crate. Defined in
| `ENABLE_METRICS` | `false` | Expose `/metrics` endpoint |
| `SERVICE_MODE` | `all` | Fleet mode: `flags`, `definitions`, or `all` |

### Property matching rollout

Exact property matching is selected per team through `TeamFeatureFlagsConfig.property_matching_version`. Version `1` preserves legacy boolean coercion, while version `2` uses exact scalar and array equality. Team metadata and local-evaluation responses expose the selected version.

### PostgreSQL

| Variable | Default | Purpose |
Expand Down
27 changes: 23 additions & 4 deletions posthog/api/cohort.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,11 @@

from posthog.api.forbid_destroy_model import ForbidDestroyModel
from posthog.api.routing import TeamAndOrgViewSetMixin
from posthog.api.services.flags_service import FlagVersionConflictError, batch_evaluate_flag_for_team
from posthog.api.services.flags_service import (
FlagVersionConflictError,
PropertyMatchingVersionConflictError,
batch_evaluate_flag_for_team,
)
from posthog.api.shared import SearchMatchTypeSerializerMixin, UserBasicSerializer
from posthog.api.utils import action
from posthog.cdp.filters import build_behavioral_event_expr
Expand Down Expand Up @@ -102,6 +106,10 @@
)
from products.cohorts.backend.models.validation import CohortTypeValidationSerializer
from products.feature_flags.backend.models.feature_flag import FeatureFlag
from products.feature_flags.backend.models.team_feature_flags_config import (
PropertyMatchingVersion,
TeamFeatureFlagsConfig,
)
from products.product_analytics.backend.facade.models import Insight


Expand Down Expand Up @@ -193,7 +201,7 @@
condition_hash: str | None = None


def generate_cohort_filter_bytecode(filter_data: dict, team: Team) -> CohortFilterBytecodeResult:

Check warning on line 204 in posthog/api/cohort.py

View workflow job for this annotation

GitHub Actions / Python code quality (depot-ubuntu-24.04)

lint:complexity

`generate_cohort_filter_bytecode` has cyclomatic complexity 11 (warn >10)

Check warning on line 204 in posthog/api/cohort.py

View workflow job for this annotation

GitHub Actions / Python code quality (depot-ubuntu-24.04)

`generate_cohort_filter_bytecode` has cyclomatic complexity 11 (warn >10)
"""
Generate HogQL bytecode for cohort filter data.
Similar to generate_template_bytecode in validation.py but for cohort-specific filters.
Expand Down Expand Up @@ -1257,7 +1265,7 @@
code="behavioral_cohort_found",
)

def update(self, cohort: Cohort, validated_data: dict, *args: Any, **kwargs: Any) -> Cohort: # type: ignore

Check warning on line 1268 in posthog/api/cohort.py

View workflow job for this annotation

GitHub Actions / Python code quality (depot-ubuntu-24.04)

lint:complexity

`update` has cyclomatic complexity 24 (warn >10)

Check warning on line 1268 in posthog/api/cohort.py

View workflow job for this annotation

GitHub Actions / Python code quality (depot-ubuntu-24.04)

`update` has cyclomatic complexity 24 (warn >10)
request = self.context["request"]
existing_has_criteria = cohort_filters_have_values(cohort.filters)
filters_changed = "filters" in validated_data and validated_data.get("filters") != cohort.filters
Expand Down Expand Up @@ -2145,6 +2153,7 @@
project_id: int,
flag_key: str,
expected_version: int,
expected_property_matching_version: int,
cursor: int,
limit: int,
) -> dict[str, Any]:
Expand All @@ -2159,11 +2168,12 @@
project_id=project_id,
flag_key=flag_key,
expected_version=expected_version,
expected_property_matching_version=expected_property_matching_version,
cursor=cursor,
limit=limit,
)
except FlagVersionConflictError:
# Permanent: the flag changed mid-run; retrying the same page cannot help.
except (FlagVersionConflictError, PropertyMatchingVersionConflictError):
# Permanent: the pinned evaluation inputs changed; retrying the same page cannot help.
raise
except requests.RequestException as err:
if (
Expand Down Expand Up @@ -2216,6 +2226,12 @@
# service refuses to evaluate under any other, so a run can never mix two
# definitions of the flag. Nullable versions coerce to 0 on both sides.
expected_version = feature_flag.version or 0
expected_property_matching_version = (
TeamFeatureFlagsConfig.objects.filter(team_id=team_id)
.values_list("property_matching_version", flat=True)
.first()
or PropertyMatchingVersion.LEGACY
)

started_at = timezone.now()
start_monotonic = time.monotonic()
Expand All @@ -2229,6 +2245,7 @@
project_id=project_id,
flag_key=feature_flag.key,
expected_version=expected_version,
expected_property_matching_version=expected_property_matching_version,
cursor=cursor,
limit=batchsize,
)
Expand Down Expand Up @@ -2281,7 +2298,9 @@
)
capture_exception(err, additional_properties={"cohort_id": cohort_id, "team_id": team_id})
error_code = (
CohortErrorCode.FLAG_CHANGED if isinstance(err, FlagVersionConflictError) else CohortErrorCode.UNKNOWN
CohortErrorCode.FLAG_CHANGED
if isinstance(err, (FlagVersionConflictError, PropertyMatchingVersionConflictError))
else CohortErrorCode.UNKNOWN
)
COHORT_FLAG_GENERATION_COMPLETED_COUNTER.labels(outcome=error_code.value).inc()
COHORT_FLAG_GENERATION_DURATION_SECONDS.labels(outcome=error_code.value).observe(
Expand Down
24 changes: 22 additions & 2 deletions posthog/api/services/flags_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,11 @@
"""The feature flag changed while a static cohort was being generated from it."""


class PropertyMatchingVersionConflictError(Exception):
"""Property matching semantics changed during static cohort generation."""


def get_flags_from_service(

Check warning on line 41 in posthog/api/services/flags_service.py

View workflow job for this annotation

GitHub Actions / Python code quality (depot-ubuntu-24.04)

lint:complexity

`get_flags_from_service` has cyclomatic complexity 12 (warn >10)

Check warning on line 41 in posthog/api/services/flags_service.py

View workflow job for this annotation

GitHub Actions / Python code quality (depot-ubuntu-24.04)

`get_flags_from_service` has cyclomatic complexity 12 (warn >10)
token: str,
distinct_id: str,
groups: dict[str, Any] | None = None,
Expand Down Expand Up @@ -155,6 +159,7 @@
expected_version: int,
cursor: int,
limit: int,
expected_property_matching_version: int = 1,
) -> dict[str, Any]:
"""
Request one page of batch flag evaluation from the Rust flags service.
Expand All @@ -172,13 +177,15 @@
limit: Page size; the service rejects (400) any value above its configured
maximum (BATCH_FLAG_EVAL_MAX_LIMIT, default 10000) rather than clamping it,
so keep this at or below that maximum
expected_property_matching_version: Matching semantics pinned for every page in the run

Returns:
The page as a dict: {"matched_person_uuids": [...], "next_cursor": int | None, "errors_count": int}
The page as a dict with matched_person_uuids, next_cursor, errors_count, and property_matching_version

Raises:
ImproperlyConfigured: `INTERNAL_REQUEST_TOKEN` is not set
FlagVersionConflictError: The flag's version no longer matches `expected_version`
PropertyMatchingVersionConflictError: The service cannot use the pinned matching version
requests.RequestException: If the HTTP request fails (timeout, connection error, etc.)
requests.HTTPError: If the service returns a non-2xx status code
"""
Expand All @@ -198,16 +205,29 @@
"project_id": project_id,
"flag_key": flag_key,
"expected_version": expected_version,
"expected_property_matching_version": expected_property_matching_version,
"cursor": cursor,
"limit": limit,
},
headers=headers,
timeout=BATCH_FLAG_EVALUATION_TIMEOUT_SECONDS,
)
if response.status_code == 409:
error = response.json().get("error")
if error == "property_matching_version_conflict":
raise PropertyMatchingVersionConflictError(
f"Property matching semantics changed while generating a cohort from feature flag '{flag_key}'. "
"Re-run cohort generation with the current matching version."
)
raise FlagVersionConflictError(
f"Feature flag '{flag_key}' changed while generating the cohort (expected version "
f"{expected_version}). Re-run cohort generation to use the latest flag definition."
)
response.raise_for_status()
return response.json()
response_data = response.json()
if response_data.get("property_matching_version", 1) != expected_property_matching_version:
raise PropertyMatchingVersionConflictError(
f"Feature flags service did not use property matching version {expected_property_matching_version} "
f"while generating a cohort from feature flag '{flag_key}'."
)
return response_data
62 changes: 61 additions & 1 deletion posthog/api/test/test_flags_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@

from posthog.api.services.flags_service import (
FlagVersionConflictError,
PropertyMatchingVersionConflictError,
batch_evaluate_flag_for_team,
get_flags_from_service,
)
Expand All @@ -34,6 +35,7 @@ def test_missing_internal_request_token_fails_fast(self, _name, token, mock_post
def test_409_maps_to_flag_version_conflict_before_raise_for_status(self, mock_post):
response = MagicMock()
response.status_code = 409
response.json.return_value = {"error": "version_conflict"}
mock_post.return_value = response

with self.assertRaises(FlagVersionConflictError):
Expand All @@ -48,7 +50,65 @@ def test_409_maps_to_flag_version_conflict_before_raise_for_status(self, mock_po
# The 409 branch must short-circuit: a generic HTTPError from raise_for_status
# would lose the version-conflict semantics the paging loop relies on.
response.raise_for_status.assert_not_called()
response.json.assert_not_called()

@override_settings(INTERNAL_REQUEST_TOKEN="secret")
@patch("posthog.api.services.flags_service._FLAGS_SERVICE_SESSION.post")
def test_property_matching_version_conflict_is_permanent(self, mock_post):
response = MagicMock()
response.status_code = 409
response.json.return_value = {"error": "property_matching_version_conflict"}
mock_post.return_value = response

with self.assertRaises(PropertyMatchingVersionConflictError):
batch_evaluate_flag_for_team(
team_id=1,
project_id=1,
flag_key="my-flag",
expected_version=7,
expected_property_matching_version=2,
cursor=0,
limit=100,
)

@override_settings(INTERNAL_REQUEST_TOKEN="secret")
@patch("posthog.api.services.flags_service._FLAGS_SERVICE_SESSION.post")
def test_accepts_legacy_response_without_property_matching_version(self, mock_post):
response_data: dict[str, object] = {"matched_person_uuids": [], "next_cursor": None, "errors_count": 0}
response = MagicMock()
response.status_code = 200
response.json.return_value = response_data
mock_post.return_value = response

result = batch_evaluate_flag_for_team(
team_id=1,
project_id=1,
flag_key="my-flag",
expected_version=7,
expected_property_matching_version=1,
cursor=0,
limit=100,
)

self.assertEqual(result, response_data)

@override_settings(INTERNAL_REQUEST_TOKEN="secret")
@patch("posthog.api.services.flags_service._FLAGS_SERVICE_SESSION.post")
def test_rejects_response_without_pinned_property_matching_version(self, mock_post):
response = MagicMock()
response.status_code = 200
response.json.return_value = {"matched_person_uuids": [], "next_cursor": None, "errors_count": 0}
mock_post.return_value = response

with self.assertRaises(PropertyMatchingVersionConflictError):
batch_evaluate_flag_for_team(
team_id=1,
project_id=1,
flag_key="my-flag",
expected_version=7,
expected_property_matching_version=2,
cursor=0,
limit=100,
)

@override_settings(INTERNAL_REQUEST_TOKEN="secret")
@patch("posthog.api.services.flags_service._FLAGS_SERVICE_SESSION.post")
Expand Down
48 changes: 34 additions & 14 deletions posthog/storage/team_metadata_cache.py
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,10 @@
get_cache_stats as get_cache_stats_generic,
)

from products.feature_flags.backend.models.team_feature_flags_config import TeamFeatureFlagsConfig
from products.feature_flags.backend.models.team_feature_flags_config import (
PropertyMatchingVersion,
TeamFeatureFlagsConfig,
)

logger = structlog.get_logger(__name__)

Expand Down Expand Up @@ -156,14 +159,20 @@ def _serialize_team_field(field: str, value: Any) -> Any:
return value


def _serialize_team_to_metadata(team: Team, minimal_flag_called_events: bool | None = None) -> dict[str, Any]:
def _serialize_team_to_metadata(
team: Team,
minimal_flag_called_events: bool | None = None,
property_matching_version: int | None = None,
) -> dict[str, Any]:
"""
Serialize a Team object to metadata dictionary.

Args:
team: Team object with organization and project already loaded
minimal_flag_called_events: Pre-fetched value from TeamFeatureFlagsConfig, to let
batch callers avoid an N+1 query. If None, this looks it up itself.
property_matching_version: Pre-fetched property matching semantics version. If None,
this looks it up with the other feature flags config value.

Returns:
Dictionary containing full team metadata
Expand All @@ -176,19 +185,22 @@ def _serialize_team_to_metadata(team: Team, minimal_flag_called_events: bool | N
metadata["organization_name"] = team.organization.name if team.organization else None
metadata["project_name"] = team.project.name if team.project else None

if minimal_flag_called_events is None:
if minimal_flag_called_events is None or property_matching_version is None:
# Deliberately not team.teamfeatureflagsconfig: the reverse O2O accessor gets
# cache-populated as a side effect of register_team_extension_signal's
# get_or_create(team=instance, ...) at team-creation time, so an already-loaded
# Team object can carry a stale cached value if the config row changes after
# the Team was loaded. An explicit query is always fresh.
minimal_flag_called_events = (
config_values = (
TeamFeatureFlagsConfig.objects.filter(team_id=team.id)
.values_list("minimal_flag_called_events", flat=True)
.values_list("minimal_flag_called_events", "property_matching_version")
.first()
or False
)
metadata["minimal_flag_called_events"] = minimal_flag_called_events
if config_values is not None:
minimal_flag_called_events = config_values[0]
property_matching_version = config_values[1]
metadata["minimal_flag_called_events"] = minimal_flag_called_events or False
metadata["property_matching_version"] = property_matching_version or PropertyMatchingVersion.LEGACY

return metadata

Expand All @@ -207,14 +219,17 @@ def _batch_load_team_metadata(teams: list[Team]) -> dict[int, dict[str, Any]]:
Returns:
Dict mapping team_id -> metadata dict
"""
minimal_flag_called_events_by_team = dict(
TeamFeatureFlagsConfig.objects.filter(team_id__in=[team.id for team in teams]).values_list(
"team_id", "minimal_flag_called_events"
)
)
config_by_team_id = {
team_id: (minimal_flag_called_events, property_matching_version)
for team_id, minimal_flag_called_events, property_matching_version in TeamFeatureFlagsConfig.objects.filter(
team_id__in=[team.id for team in teams]
).values_list("team_id", "minimal_flag_called_events", "property_matching_version")
}
return {
team.id: _serialize_team_to_metadata(
team, minimal_flag_called_events=minimal_flag_called_events_by_team.get(team.id, False)
team,
minimal_flag_called_events=config_by_team_id.get(team.id, (False, PropertyMatchingVersion.LEGACY))[0],
property_matching_version=config_by_team_id.get(team.id, (False, PropertyMatchingVersion.LEGACY))[1],
)
for team in teams
}
Expand Down Expand Up @@ -332,7 +347,12 @@ def verify_team_metadata(

# Compare only fields we care about (defined in TEAM_METADATA_FIELDS + derived fields).
# This allows removing fields from the cache without triggering unnecessary fixes.
fields_to_check = set(TEAM_METADATA_FIELDS) | {"organization_name", "project_name", "minimal_flag_called_events"}
fields_to_check = set(TEAM_METADATA_FIELDS) | {
"organization_name",
"project_name",
"minimal_flag_called_events",
"property_matching_version",
}
diffs = []
for key in fields_to_check:
db_val = db_data.get(key)
Expand Down
Loading
Loading