# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-redisclusterfailovererror-is-imminent-the-slot-ownership-divergence-window.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-redisclusterfailovererror-is-imminent-the-slot-ownership-divergence-window.warrant.json
# Markdown SHA-256:   39a0a40e2615092f1905a6088c00bbc8ab9556a3db2999902731cdb2292e5ed2
# Code AST SHA-256:   4bf38e5d8605ec8a3c04ec07deab7ab2c82c50147f511a80a19d58a74a2a360b
# Verified Classes:   RedisClusterFailoverError, SlotView
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import time
from dataclasses import dataclass

class RedisClusterFailoverError(RuntimeError):
    """Raised when slot-ownership divergence cannot converge because the
    higher-configEpoch claimant is unreachable."""

@dataclass
class SlotView:
    owner_id: str
    config_epoch: int
    reachable: bool

def resolve_slot_owner(views: list[SlotView], cluster_node_timeout_ms: int) -> str:
    """Return the converged owner, or raise RedisClusterFailoverError.

    `views` are independent observations of the same slot from different
    cluster nodes. A redirect cycle across these views is expected during
    promotion; we only raise when the claimant is unreachable.
    """
    if not views:
        raise RedisClusterFailoverError("no slot views available; cannot resolve owner")

    claimant = max(views, key=lambda v: v.config_epoch)
    disagreeing = [v for v in views if v.owner_id != claimant.owner_id]

    if not disagreeing:
        return claimant.owner_id

    if not claimant.reachable:
        raise RedisClusterFailoverError(
            f"slot divergence with unreachable higher-epoch claimant "
            f"{claimant.owner_id} (epoch={claimant.config_epoch}); "
            f"cannot converge within {cluster_node_timeout_ms}ms"
        )

    # Divergence is convergence-in-progress: back off, do not raise.
    time.sleep(min(0.05, cluster_node_timeout_ms / 1000.0))
    return claimant.owner_id