diff --git a/src/sentry/issues/derived/aggregators.py b/src/sentry/issues/derived/aggregators.py index 8d05ad129694..0e64239345c4 100644 --- a/src/sentry/issues/derived/aggregators.py +++ b/src/sentry/issues/derived/aggregators.py @@ -43,6 +43,7 @@ from sentry.issues.derived.framework import ( Aggregator, AggregatorResult, + Scope, StateView, aggregator, emit, @@ -157,7 +158,6 @@ def track_root_cause(state: StateView, entry: GroupActionLogEntry) -> Aggregator @aggregator( (HAS_OPEN_FIX_PR,), - deps=(STATUS,), scope=( ResolvedInPullRequestAction, PullRequestClosedAction, @@ -261,6 +261,7 @@ def track_last_completed_autofix_step( @aggregator( (BLOCKER,), deps=(STATUS, HAS_OPEN_FIX_PR, LAST_COMPLETED_AUTOFIX_STEP), + scope=Scope.DEPS, ) def track_blocker(state: StateView, entry: GroupActionLogEntry) -> AggregatorResult: """Track the human action blocking the issue's progress toward resolution. diff --git a/src/sentry/issues/derived/framework.py b/src/sentry/issues/derived/framework.py index 4e20d28e0001..3263244c7bac 100644 --- a/src/sentry/issues/derived/framework.py +++ b/src/sentry/issues/derived/framework.py @@ -10,7 +10,7 @@ from dataclasses import dataclass from datetime import datetime from enum import IntEnum, StrEnum -from typing import Any, ClassVar, Final, Protocol, runtime_checkable +from typing import Any, ClassVar, Final, Literal, Protocol, runtime_checkable _MISSING = object() @@ -281,6 +281,19 @@ def emit(*entries: FeatureEntry) -> AggregatorResult: return StateUpdate(dict(entries)) +class Scope(StrEnum): + """Special aggregator scopes. + + ALL runs for every entry. DEPS runs whenever any dependency producer runs. + """ + + ALL = "all" + DEPS = "deps" + + +type EffectiveScope = Literal[Scope.ALL] | frozenset[int] + + @dataclass(frozen=True) class Aggregator[E: HasType]: """A named function that reads from dep features and writes to output features.""" @@ -289,7 +302,7 @@ class Aggregator[E: HasType]: deps: tuple[Feature[Any], ...] outputs: tuple[Feature[Any], ...] fn: AggregatorFn[E] - scope: tuple[int, ...] | None = None + scope: Scope | tuple[int, ...] = Scope.ALL class _HasGetType(Protocol): @@ -310,12 +323,14 @@ def aggregator[E: HasType]( outputs: tuple[Feature[Any], ...], *, deps: tuple[Feature[Any], ...] = (), - scope: tuple[ScopeItem, ...] | None = None, + scope: Scope | tuple[ScopeItem, ...] = Scope.ALL, ) -> Callable[[AggregatorFn[E]], Aggregator[E]]: - """Decorator to create an Aggregator. `scope` accepts enum members or classes with get_type().""" + """Create an aggregator with a fixed, universal, or dependency-derived scope.""" if not outputs: raise ValueError("aggregator must declare at least one output") - raw_scope = tuple(_scope_int(s) for s in scope) if scope is not None else None + if scope is Scope.DEPS and not deps: + raise ValueError("Scope.DEPS requires at least one dependency") + raw_scope = scope if isinstance(scope, Scope) else tuple(_scope_int(s) for s in scope) def decorator(fn: AggregatorFn[E]) -> Aggregator[E]: return Aggregator(name=fn.__name__, deps=deps, outputs=outputs, fn=fn, scope=raw_scope) @@ -323,6 +338,13 @@ def decorator(fn: AggregatorFn[E]) -> Aggregator[E]: return decorator +@dataclass(frozen=True) +class _ValidatedPipeline[E: HasType]: + aggregators: tuple[Aggregator[E], ...] + features: tuple[Feature[Any], ...] + effective_scopes: dict[str, EffectiveScope] + + # --------------------------------------------------------------------------- # Pipeline # --------------------------------------------------------------------------- @@ -344,9 +366,16 @@ def __init__( ) -> None: self._check_mutations = check_mutations aggregators = tuple(aggregators) - self._aggregators, self._features = _validate_and_sort(aggregators) + validated = _validate_and_sort(aggregators) + self._aggregators = validated.aggregators + self._features = validated.features self._steps = tuple( - (agg, frozenset({*agg.deps, *agg.outputs}), frozenset(agg.outputs)) + ( + agg, + frozenset({*agg.deps, *agg.outputs}), + frozenset(agg.outputs), + validated.effective_scopes[agg.name], + ) for agg in self._aggregators ) payload = f"{self._version}:" + ",".join(sorted(f.content_id for f in self._features)) @@ -371,9 +400,10 @@ def initial_state(self) -> State: def step(self, state: State, entry: E) -> State: entry_type = entry.type - for agg, view_fields, output_fields in self._steps: - if agg.scope is not None and entry_type not in agg.scope: - continue + for agg, view_fields, output_fields, effective_scope in self._steps: + if effective_scope is not Scope.ALL: + if entry_type not in effective_scope: + continue subset = state.view(view_fields) snapshot = copy.deepcopy(subset._data) if self._check_mutations else None result = agg.fn(subset, entry) @@ -442,7 +472,7 @@ def _ensure_no_aliasing(features: Iterable[Feature[Any]]) -> tuple[Feature[Any], def _validate_and_sort[E: HasType]( aggregators: tuple[Aggregator[E], ...], -) -> tuple[tuple[Aggregator[E], ...], tuple[Feature[Any], ...]]: +) -> _ValidatedPipeline[E]: output_owners: dict[str, Aggregator[E]] = {} for agg in aggregators: for feature in agg.outputs: @@ -488,6 +518,43 @@ def _validate_and_sort[E: HasType]( remaining = {a.name for a in aggregators} - {a.name for a in order} raise ValueError(f"Cycle detected among aggregators: {remaining}") + effective_scopes: dict[str, EffectiveScope] = {} + for agg in order: + if agg.scope is Scope.DEPS: + if not agg.deps: + raise ValueError(f"Aggregator {agg.name!r} uses Scope.DEPS without dependencies") + scope_values: set[int] = set() + for dep in agg.deps: + producer = output_owners[dep.name] + if producer.name == agg.name: + raise ValueError( + f"Aggregator {agg.name!r} cannot derive its scope from its own output" + ) + producer_scope = effective_scopes[producer.name] + if producer_scope is Scope.ALL: + effective_scopes[agg.name] = Scope.ALL + break + scope_values.update(producer_scope) + else: + effective_scopes[agg.name] = frozenset(scope_values) + elif agg.scope is Scope.ALL: + effective_scopes[agg.name] = Scope.ALL + else: + effective_scopes[agg.name] = frozenset(agg.scope) + + for agg in aggregators: + scope = effective_scopes[agg.name] + for dep in agg.deps: + producer = output_owners[dep.name] + producer_scope = effective_scopes[producer.name] + if scope is Scope.ALL: + continue + if producer_scope is Scope.ALL or not scope.issuperset(producer_scope): + raise ValueError( + f"Aggregator {agg.name!r} has a scope that does not cover " + f"dependency {dep.name!r} produced by {producer.name!r}" + ) + all_features = _ensure_no_aliasing(f for agg in aggregators for f in (*agg.deps, *agg.outputs)) - return tuple(order), all_features + return _ValidatedPipeline(tuple(order), all_features, effective_scopes) diff --git a/tests/sentry/issues/derived/test_framework.py b/tests/sentry/issues/derived/test_framework.py index 9c825d978b14..51634bf54e6b 100644 --- a/tests/sentry/issues/derived/test_framework.py +++ b/tests/sentry/issues/derived/test_framework.py @@ -1,4 +1,5 @@ from datetime import datetime, timezone +from enum import IntEnum import pytest @@ -10,6 +11,7 @@ Feature, OptionalCodec, Pipeline, + Scope, State, StateUpdate, StateView, @@ -18,6 +20,12 @@ from sentry.issues.progress_state import IssueProgressState +class EntryType(IntEnum): + FIRST = 1 + SECOND = 2 + THIRD = 3 + + def test_mutation_checking_catches_in_place_mutation() -> None: ITEMS = Feature[list[str]]("items", default_factory=list) @@ -49,6 +57,157 @@ def test_state_updated_tracks_merged_features() -> None: assert state[B] == 0 +def test_dependency_scope_must_cover_producer_scope() -> None: + A = Feature[int]("a", default=0) + B = Feature[int]("b", default=0) + + @aggregator((A,), scope=(EntryType.FIRST, EntryType.SECOND)) + def produce_a(state: StateView, entry: object) -> AggregatorResult: + return None + + @aggregator((B,), deps=(A,), scope=(EntryType.FIRST,)) + def use_a(state: StateView, entry: object) -> AggregatorResult: + return None + + with pytest.raises(ValueError, match="scope that does not cover dependency 'a'"): + Pipeline([produce_a, use_a]) + + +def test_scoped_aggregator_cannot_depend_on_all_scope_aggregator() -> None: + A = Feature[int]("a", default=0) + B = Feature[int]("b", default=0) + + @aggregator((A,)) + def produce_a(state: StateView, entry: object) -> AggregatorResult: + return None + + @aggregator((B,), deps=(A,), scope=(EntryType.FIRST,)) + def use_a(state: StateView, entry: object) -> AggregatorResult: + return None + + with pytest.raises(ValueError, match="scope that does not cover dependency 'a'"): + Pipeline([produce_a, use_a]) + + +def test_dependency_scope_can_be_a_superset_of_producer_scope() -> None: + A = Feature[int]("a", default=0) + B = Feature[int]("b", default=0) + + @aggregator((A,), scope=(EntryType.FIRST,)) + def produce_a(state: StateView, entry: object) -> AggregatorResult: + return None + + @aggregator((B,), deps=(A,), scope=(EntryType.FIRST, EntryType.SECOND)) + def use_a(state: StateView, entry: object) -> AggregatorResult: + return None + + assert Pipeline([produce_a, use_a]).aggregators == (produce_a, use_a) + + +def test_all_scope_aggregator_covers_scoped_dependency() -> None: + A = Feature[int]("a", default=0) + B = Feature[int]("b", default=0) + + @aggregator((A,), scope=(EntryType.FIRST,)) + def produce_a(state: StateView, entry: object) -> AggregatorResult: + return None + + @aggregator((B,), deps=(A,)) + def use_a(state: StateView, entry: object) -> AggregatorResult: + return None + + assert Pipeline([produce_a, use_a]).aggregators == (produce_a, use_a) + + +def test_default_scope_is_all() -> None: + A = Feature[int]("a", default=0) + + @aggregator((A,)) + def produce_a(state: StateView, entry: object) -> AggregatorResult: + return None + + assert produce_a.scope is Scope.ALL + + +def test_deps_scope_runs_for_union_of_dependency_scopes() -> None: + A = Feature[int]("a", default=0) + B = Feature[int]("b", default=0) + C = Feature[int]("c", default=0) + + @aggregator((A,), scope=(EntryType.FIRST,)) + def produce_a(state: StateView, entry: object) -> AggregatorResult: + return StateUpdate({A: state[A] + 1}) + + @aggregator((B,), scope=(EntryType.SECOND,)) + def produce_b(state: StateView, entry: object) -> AggregatorResult: + return StateUpdate({B: state[B] + 1}) + + @aggregator((C,), deps=(A, B), scope=Scope.DEPS) + def use_deps(state: StateView, entry: object) -> AggregatorResult: + return StateUpdate({C: state[C] + 1}) + + class Entry: + def __init__(self, type: EntryType) -> None: + self.type = type + + state = Pipeline([produce_a, produce_b, use_deps]).run( + [Entry(EntryType.FIRST), Entry(EntryType.SECOND), Entry(EntryType.THIRD)] + ) + + assert state[A] == 1 + assert state[B] == 1 + assert state[C] == 2 + + +def test_deps_scope_is_resolved_transitively() -> None: + A = Feature[int]("a", default=0) + B = Feature[int]("b", default=0) + C = Feature[int]("c", default=0) + + @aggregator((A,), scope=(EntryType.FIRST,)) + def produce_a(state: StateView, entry: object) -> AggregatorResult: + return None + + @aggregator((B,), deps=(A,), scope=Scope.DEPS) + def use_a(state: StateView, entry: object) -> AggregatorResult: + return None + + @aggregator((C,), deps=(B,), scope=(EntryType.SECOND,)) + def use_b(state: StateView, entry: object) -> AggregatorResult: + return None + + with pytest.raises(ValueError, match="scope that does not cover dependency 'b'"): + Pipeline([produce_a, use_a, use_b]) + + +def test_deps_scope_resolves_to_all_when_dependency_scope_is_all() -> None: + A = Feature[int]("a", default=0) + B = Feature[int]("b", default=0) + + @aggregator((A,)) + def produce_a(state: StateView, entry: object) -> AggregatorResult: + return StateUpdate({A: state[A] + 1}) + + @aggregator((B,), deps=(A,), scope=Scope.DEPS) + def use_a(state: StateView, entry: object) -> AggregatorResult: + return StateUpdate({B: state[B] + 1}) + + class Entry: + type = EntryType.THIRD + + state = Pipeline([produce_a, use_a]).run([Entry()]) + + assert state[A] == 1 + assert state[B] == 1 + + +def test_deps_scope_requires_dependency() -> None: + A = Feature[int]("a", default=0) + + with pytest.raises(ValueError, match="requires at least one dependency"): + aggregator((A,), scope=Scope.DEPS) + + class TestDateTimeCodec: def test_json_round_trip(self) -> None: codec = DateTimeCodec()