# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/through-what-quorum-and-fencing-token-gates-must-an-instance-coordinate-distributed-leadership-to-prevent-split-brain-writes-and-zombie-master-mutations.md
# Remnant Warrant:    https://corpusfetch.com/c/through-what-quorum-and-fencing-token-gates-must-an-instance-coordinate-distributed-leadership-to-prevent-split-brain-writes-and-zombie-master-mutations.warrant.json
# Markdown SHA-256:   f5b7bdaa7492846a94bb161cf6bccec6283bfe11a352dbf2e2e377add7d33ebb
# Code AST SHA-256:   330b2216c8b68ebb842322135a1c28d735f62b480f0a600a561c75fa728c29b5
# Verified Classes:   StaleFencingTokenError, QuorumLostError, FencedLease, FencedCoordinatorAndStore
# Defensive Guards:   6 fail-closed invariants (zero external deps)

import sqlite3
import time
from dataclasses import dataclass
from typing import Any


class StaleFencingTokenError(PermissionError):
    '''Raised when a storage target rejects a write from a superseded zombie leader.'''
    pass


class QuorumLostError(RuntimeError):
    '''Raised when a node cannot reach a strict majority of cluster voters.'''
    pass


@dataclass(frozen=True)
class FencedLease:
    resource_key: str
    holder_id: str
    fencing_token: int
    local_deadline_monotonic: float

    def is_locally_valid(self, safety_margin_sec: float = 0.5) -> bool:
        '''Returns True only if monotonic time is well within the conservative deadline.'''
        return (time.monotonic() + safety_margin_sec) < self.local_deadline_monotonic


def verify_strict_quorum(reachable_voters: int, total_configured_voters: int) -> None:
    '''
    Enforces the strict majority quorum invariant: reachable > total // 2.
    Rejects even-split ties (e.g., 2 out of 4 nodes) to prevent split-brain promotion.
    '''
    if total_configured_voters < 1:
        raise ValueError("total_configured_voters must be >= 1")
    required_majority = (total_configured_voters // 2) + 1
    if reachable_voters < required_majority:
        raise QuorumLostError(
            f"Split-brain guard: reachable voters ({reachable_voters}/{total_configured_voters}) "
            f"below required majority quorum of {required_majority}. Stepping down."
        )


class FencedCoordinatorAndStore:
    '''
    Demonstrates end-to-end monotonic fencing:
    1. Issues strictly increasing integer fencing tokens on every lease acquisition.
    2. Rejects any storage mutation carrying a token older than max_seen_token.
    '''

    def __init__(self, conn: sqlite3.Connection) -> None:
        self.conn = conn
        with self.conn:
            self.conn.execute(
                "CREATE TABLE IF NOT EXISTS lease_registry ("
                "  resource_key TEXT PRIMARY KEY,"
                "  holder_id TEXT NOT NULL,"
                "  fencing_token INTEGER NOT NULL,"
                "  expires_at_epoch REAL NOT NULL"
                ")"
            )
            self.conn.execute(
                "CREATE TABLE IF NOT EXISTS fenced_storage ("
                "  resource_key TEXT PRIMARY KEY,"
                "  payload TEXT NOT NULL,"
                "  max_seen_token INTEGER NOT NULL"
                ")"
            )

    def acquire_or_renew_lease(
        self,
        resource_key: str,
        holder_id: str,
        ttl_sec: float,
        *,
        last_progress_monotonic: float,
        max_stall_sec: float = 15.0,
    ) -> FencedLease:
        '''
        Acquires a new lease (incrementing fencing_token) or renews an existing lease
        ONLY if the worker's main loop has recorded recent forward progress.
        '''
        # Gate 1: Refuse renewal if the worker's main execution loop is stalled/deadlocked
        stall_duration = time.monotonic() - last_progress_monotonic
        if stall_duration > max_stall_sec:
            raise RuntimeError(
                f"Refusing lease renewal for {holder_id!r}: main loop stalled for {stall_duration:.1f}s"
            )

        t_start = time.monotonic()
        now_epoch = time.time()

        with self.conn:
            row = self.conn.execute(
                "SELECT holder_id, fencing_token, expires_at_epoch "
                "FROM lease_registry WHERE resource_key = ?",
                (resource_key,),
            ).fetchone()

            if row is None:
                token = 1
                self.conn.execute(
                    "INSERT INTO lease_registry (resource_key, holder_id, fencing_token, expires_at_epoch) "
                    "VALUES (?, ?, ?, ?)",
                    (resource_key, holder_id, token, now_epoch + ttl_sec),
                )
            else:
                curr_holder, curr_token, expires_at = row[0], int(row[1]), float(row[2])
                if curr_holder != holder_id and expires_at > now_epoch:
                    raise PermissionError(
                        f"Resource {resource_key!r} is actively leased by {curr_holder!r}"
                    )
                # Increment monotonic fencing token on every leadership transition or renewal
                token = curr_token + 1
                self.conn.execute(
                    "UPDATE lease_registry "
                    "SET holder_id = ?, fencing_token = ?, expires_at_epoch = ? "
                    "WHERE resource_key = ?",
                    (holder_id, token, now_epoch + ttl_sec, resource_key),
                )

        return FencedLease(
            resource_key=resource_key,
            holder_id=holder_id,
            fencing_token=token,
            local_deadline_monotonic=t_start + ttl_sec,
        )

    def fenced_write(self, lease: FencedLease, payload: str) -> int:
        '''
        Executes a storage write strictly guarded by `lease.fencing_token`.
        Even if a paused zombie leader bypasses local checks, the storage layer
        rejects any token smaller than `max_seen_token`.
        '''
        if not lease.is_locally_valid():
            raise StaleFencingTokenError(
                f"Local lease deadline expired for {lease.holder_id!r} (token={lease.fencing_token})"
            )

        with self.conn:
            row = self.conn.execute(
                "SELECT max_seen_token FROM fenced_storage WHERE resource_key = ?",
                (lease.resource_key,),
            ).fetchone()

            if row is None:
                self.conn.execute(
                    "INSERT INTO fenced_storage (resource_key, payload, max_seen_token) "
                    "VALUES (?, ?, ?)",
                    (lease.resource_key, payload, lease.fencing_token),
                )
                return lease.fencing_token

            max_seen = int(row[0])
            # Gate 2: Authoritative Storage-Side Fencing Check
            if lease.fencing_token < max_seen:
                raise StaleFencingTokenError(
                    f"Zombie write blocked! Leader {lease.holder_id!r} presented stale "
                    f"fencing_token={lease.fencing_token}, but storage is already at token={max_seen}."
                )

            self.conn.execute(
                "UPDATE fenced_storage "
                "SET payload = ?, max_seen_token = ? "
                "WHERE resource_key = ? AND max_seen_token <= ?",
                (payload, lease.fencing_token, lease.resource_key, lease.fencing_token),
            )
            return lease.fencing_token