Source code for eclypse.policies.failure.latency_spike

"""Latency spike policy."""

from __future__ import annotations

from typing import TYPE_CHECKING

from eclypse.policies._filters import (
    clamp,
    coerce_numeric_like,
    ensure_numeric_value,
    iter_selected_edges,
)
from eclypse.policies._helpers import validate_probability
from eclypse.utils.constants import MIN_LATENCY

if TYPE_CHECKING:
    from eclypse.graph.asset_graph import AssetGraph
    from eclypse.policies._filters import EdgeFilter
    from eclypse.utils.types import (
        NumericBasis,
        UpdatePolicy,
    )




def latency_spike(
    probability: float,
    *,
    min_increase: float = 1.0,
    max_increase: float | None = None,
    factor: float | None = None,
    latency_key: str = "latency",
    edge_ids: list[tuple[str, str]] | None = None,
    edge_filter: EdgeFilter | None = None,
    basis: NumericBasis = "current",
) -> UpdatePolicy:
    """Inject random latency spikes on selected edges.

    Args:
        probability (float): Probability of applying a spike to each selected edge.
        min_increase (float): Minimum additive spike size when using additive mode.
        max_increase (float | None): Maximum additive spike size when using additive
            mode. Defaults to ``min_increase``.
        factor (float | None): Optional multiplicative spike factor. When provided,
            additive spike parameters are ignored.
        latency_key (str): Edge asset storing latency.
        edge_ids (list[tuple[str, str]] | None): Optional explicit list of target
            edges.
        edge_filter (EdgeFilter | None): Optional predicate to filter target edges.
        basis (NumericBasis):
            ``"current"`` compounds spikes. ``"initial"`` spikes from the first
            value seen by this policy.

    Returns:
        UpdatePolicy: A graph update policy implementing latency spikes.
    """
    validate_probability("probability", probability)
    if factor is not None and factor < 0:
        raise ValueError("factor must be non-negative.")
    if min_increase < 0:
        raise ValueError("min_increase must be non-negative.")

    spike_ceiling = min_increase if max_increase is None else max_increase
    if spike_ceiling < min_increase:
        raise ValueError("max_increase must be greater than or equal to min_increase.")

    baselines: dict[tuple[str, ...], float] = {}

    def policy(graph: AssetGraph):
        if basis not in {"current", "initial"}:
            raise ValueError('basis must be either "current" or "initial".')

        for source, target, data in iter_selected_edges(
            graph,
            edge_ids=edge_ids,
            edge_filter=edge_filter,
        ):
            current = ensure_numeric_value(latency_key, data[latency_key])
            source_value = (
                current
                if basis == "current"
                else baselines.setdefault(
                    ("edge", source, target, latency_key), current
                )
            )
            if graph.rnd.random() >= probability:
                continue

            if factor is not None:
                new_value = source_value * factor
            else:
                new_value = source_value + graph.rnd.uniform(
                    min_increase,
                    spike_ceiling,
                )

            data[latency_key] = coerce_numeric_like(
                data[latency_key],
                clamp(new_value, lower=MIN_LATENCY),
            )

        graph.logger.trace("Applied latency_spike policy.")

    return policy