# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-does-a-watermark-lagging-event-time-skew-silently-drop-streaming-join-matches.md
# Remnant Warrant:    https://corpusfetch.com/c/when-does-a-watermark-lagging-event-time-skew-silently-drop-streaming-join-matches.warrant.json
# Markdown SHA-256:   ef4d67e4034387b957825518262c27ca3e01f7384ccae5b12c6506aaae9437a1
# Code AST SHA-256:   d51e983c621831936a20f064f7d7440b504ec8bbd20b64800a690e41ceaf5ccc
# Verified Classes:   WatermarkSkewError, WatermarkGate
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass


class WatermarkSkewError(RuntimeError):
    """Raised when the configured watermark lag cannot absorb the observed event-time skew."""


@dataclass(frozen=True)
class WatermarkGate:
    watermark_lag_ms: int
    margin_ms: int = 2000

    def verify(self, skew_p99_ms: int, same_partition_set: bool = True) -> bool:
        if not same_partition_set:
            raise WatermarkSkewError(
                "skew statistic measured on a different partition set than the join input"
            )
        if self.watermark_lag_ms < skew_p99_ms + self.margin_ms:
            raise WatermarkSkewError(
                f"lag {self.watermark_lag_ms}ms < skew_p99 {skew_p99_ms}ms + margin {self.margin_ms}ms"
            )
        return True


def dispatch_join_mode(gate: WatermarkGate, skew_p99_ms: int, same_partition_set: bool = True) -> str:
    try:
        gate.verify(skew_p99_ms, same_partition_set)
    except WatermarkSkewError:
        return "outer-with-alerted-late-path"
    return "inner"