diff --git a/src/adcp/reporting/subscriptions/_clock.py b/src/adcp/reporting/subscriptions/_clock.py index b9e1b63bc..50b86d3ec 100644 --- a/src/adcp/reporting/subscriptions/_clock.py +++ b/src/adcp/reporting/subscriptions/_clock.py @@ -87,6 +87,17 @@ def __init__(self, detail: str, message: str) -> None: else "COVERAGE_INPUT_INVALID" ) + @classmethod + def budget_exceeded(cls, budget: str, message: str) -> CoverageInputError: + """``COVERAGE_BUDGET_EXCEEDED``: ``budget`` is ``max_periods`` or ``max_iterations``.""" + if budget not in {"max_periods", "max_iterations"}: + raise AssertionError(f"unknown coverage budget {budget!r}") + error = cls.__new__(cls) + super(CoverageInputError, error).__init__(message) + error.detail = budget + error.code = "COVERAGE_BUDGET_EXCEEDED" + return error + _DURATION = re.compile(r"P(?:(\d+)Y)?(?:(\d+)M)?(?:(\d+)D)?(?:T(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?)?") _SLA = re.compile(r"P(?:(\d+)D)?(?:T(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?)?") diff --git a/src/adcp/reporting/subscriptions/coverage.py b/src/adcp/reporting/subscriptions/coverage.py index 276d2371f..6664cede5 100644 --- a/src/adcp/reporting/subscriptions/coverage.py +++ b/src/adcp/reporting/subscriptions/coverage.py @@ -93,6 +93,12 @@ def _require_instant(value: Any, *, field: str) -> datetime: # -- plain-data inputs ------------------------------------------------------ +#: Owed (buy, period) pairs one coverage call may return before failing closed. +DEFAULT_MAX_PERIODS = 100_000 +#: Period boundaries one coverage call may examine before failing closed. +DEFAULT_MAX_ITERATIONS = 2_000_000 + + @dataclass(frozen=True) class SubscriptionBuyFlight: """A flight a buy had before a later change superseded it at ``superseded_at``. @@ -724,9 +730,17 @@ def expected_subscription_periods( version_configurations: Mapping[int, Any] | None = None, timezone_changes: Sequence[AccountTimezoneChange] = (), include_unresolved: bool = False, + max_periods: int = DEFAULT_MAX_PERIODS, + max_iterations: int = DEFAULT_MAX_ITERATIONS, ) -> list[ExpectedSubscriptionPeriod]: """Every period one subscription owes one account, from plain data. + Fails closed rather than return a partial set: more than ``max_periods`` + owed (buy, period) pairs, or more than ``max_iterations`` boundaries + examined, raises :class:`CoverageInputError` with code + ``COVERAGE_BUDGET_EXCEEDED`` (the TypeScript SDK's ``max_periods`` and + ``max_iterations``, with the same defaults). + :param subscription: the ``get_principal`` readback (``core/reporting-subscription-state.json``) as a model or mapping, or a sequence of :class:`SubscriptionVersionWindow`. @@ -756,6 +770,10 @@ def expected_subscription_periods( ``changed_at`` (``timezone_cutovers[].effective_at``). """ now = utc_instant(now, field="now") + for name, budget in (("max_periods", max_periods), ("max_iterations", max_iterations)): + if type(budget) is not int or budget < 1: + raise CoverageInputError("invalid_input", f"{name} must be a positive integer") + iterations = pairs = 0 windows = _windows(subscription, version_configurations) if not windows: return [] @@ -861,6 +879,14 @@ def expected_subscription_periods( if lower >= upper: continue for start, end in calendar.periods(starts_from=lower, starts_before=upper): + iterations += 1 + if iterations > max_iterations: + raise CoverageInputError.budget_exceeded( + "max_iterations", + f"Coverage would examine more than max_iterations ({max_iterations}) " + f"period boundaries for subscription {subscription_id!r} on account " + f"{account_id!r}. Narrow the input or raise max_iterations.", + ) if not include_unresolved and end > now: break flight_start, flight_stop = buy.flight_at(end) @@ -892,7 +918,16 @@ def expected_subscription_periods( expected_at, withheld, ) - entry.media_buy_ids.add(buy.media_buy_id) + if buy.media_buy_id not in entry.media_buy_ids: + entry.media_buy_ids.add(buy.media_buy_id) + pairs += 1 + if pairs > max_periods: + raise CoverageInputError.budget_exceeded( + "max_periods", + f"Subscription {subscription_id!r} owes account {account_id!r} " + f"more than max_periods ({max_periods}) periods. Narrow the " + "input or raise max_periods.", + ) periods = [ ExpectedSubscriptionPeriod( @@ -1046,6 +1081,8 @@ def _subscription_id(subscription: Any, windows: Sequence[SubscriptionVersionWin __all__ = [ "AccountTimezoneChange", + "DEFAULT_MAX_ITERATIONS", + "DEFAULT_MAX_PERIODS", "ExpectedSubscriptionPeriod", "SubscriptionAuthorization", "SubscriptionBuy", diff --git a/tests/test_reporting_subscription_coverage.py b/tests/test_reporting_subscription_coverage.py index cef1d43cd..8d24b3273 100644 --- a/tests/test_reporting_subscription_coverage.py +++ b/tests/test_reporting_subscription_coverage.py @@ -21,6 +21,7 @@ from adcp.reporting import ExpectedReportingPeriod from adcp.reporting.subscriptions import ( AccountTimezoneChange, + CoverageInputError, ExpectedSubscriptionPeriod, SubscriptionAuthorization, SubscriptionBuy, @@ -1751,3 +1752,36 @@ def test_a_bare_3_3_version_cannot_be_served_while_beta_1_is_pinned() -> None: with pytest.raises(ConfigurationError): resolve_adcp_version("3.3") assert "3.3" not in get_supported_adcp_versions() + + +# -- the budget ---------------------------------------------------------------- + + +class TestCoverageBudget: + """Coverage fails closed rather than return a partial set (COVERAGE_BUDGET_EXCEEDED).""" + + def test_more_owed_periods_than_max_periods_raise(self) -> None: + buys = [buy(f"mb-{index}", end_time="2026-10-04T04:00:00Z") for index in range(2)] + assert len(run(buys=buys, max_periods=6)) == 3 # 3 periods x 2 buys = 6 pairs + with pytest.raises(CoverageInputError) as exceeded: + run(buys=buys, max_periods=5) + error = exceeded.value + assert (error.code, error.detail, error.recovery) == ( + "COVERAGE_BUDGET_EXCEEDED", + "max_periods", + "correctable", + ) + assert error.dispatched is None and "max_periods (5)" in str(error) + + def test_more_boundaries_than_max_iterations_raise(self) -> None: + with pytest.raises(CoverageInputError) as exceeded: + run(buys=[buy(end_time="2026-10-04T04:00:00Z")], max_iterations=2) + assert (exceeded.value.code, exceeded.value.detail) == ( + "COVERAGE_BUDGET_EXCEEDED", + "max_iterations", + ) + + @pytest.mark.parametrize("name", ["max_periods", "max_iterations"]) + def test_a_budget_must_be_positive(self, name: str) -> None: + with pytest.raises(CoverageInputError, match=name): + run(buys=[buy()], **{name: 0})