Consistency Models.

# Read-Your-Writes Implementation

from dataclasses import dataclass
from datetime import datetime
from typing import Optional, Any
import asyncio
import time


@dataclass
class WriteRecord:
    """Track a client's last write."""
    key: str
    timestamp: float
    replica_id: str


class ReadYourWritesClient:
    """
    Client that guarantees read-your-writes consistency.
    
    Tracks writes and ensures subsequent reads see them.
    """
    
    def __init__(self, replicas: list, session_id: str):
        self.replicas = replicas
        self.session_id = session_id
        self.last_writes: dict[str, WriteRecord] = {}
    
    async def write(self, key: str, value: Any) -> bool:
        """Write to primary and track the write."""
        primary = self._get_primary()
        
        timestamp = time.time()
        success = await primary.write(key, value, timestamp)
        
        if success:
            # Track this write for read-your-writes
            self.last_writes[key] = WriteRecord(
                key=key,
                timestamp=timestamp,
                replica_id=primary.id
            )
        
        return success
    
    async def read(self, key: str) -> Any:
        """Read with read-your-writes guarantee."""
        last_write = self.last_writes.get(key)
        
        if last_write:
            # We wrote this key - ensure we see our write
            return await self._read_with_min_version(key, last_write.timestamp)
        else:
            # We didn't write this key - any replica is fine
            return await self._read_from_any(key)
    
    async def _read_with_min_version(self, key: str, min_timestamp: float) -> Any:
        """Read ensuring we see at least version min_timestamp."""
        
        # Strategy 1: Read from the replica we wrote to
        for replica in self.replicas:
            if replica.id == self.last_writes[key].replica_id:
                return await replica.read(key)
        
        # Strategy 2: Query replicas until we find one with our version
        for replica in self.replicas:
            value, timestamp = await replica.read_with_version(key)
            if timestamp >= min_timestamp:
                return value
        
        # Strategy 3: Wait for replication (with timeout)
        return await self._wait_for_replication(key, min_timestamp)
    
    async def _wait_for_replication(
        self, 
        key: str, 
        min_timestamp: float,
        timeout: float = 5.0
    ) -> Any:
        """Wait for a replica to catch up."""
        deadline = time.time() + timeout
        
        while time.time() < deadline:
            for replica in self.replicas:
                value, timestamp = await replica.read_with_version(key)
                if timestamp >= min_timestamp:
                    return value
            
            await asyncio.sleep(0.1)
        
        raise TimeoutError(f"Replica did not catch up within {timeout}s")
    
    async def _read_from_any(self, key: str) -> Any:
        """Read from any available replica."""
        for replica in self.replicas:
            try:
                return await replica.read(key)
            except Exception:
                continue
        raise Exception("All replicas unavailable")
    
    def _get_primary(self):
        """Get the primary replica for writes."""
        return self.replicas[0]  # Simplified

# Monotonic Reads Implementation

@dataclass
class VersionedValue:
    """Value with version for comparison."""
    value: Any
    version: int  # Logical clock or timestamp


class MonotonicReadsClient:
    """
    Client that guarantees monotonic reads.
    
    Once you read version N, you never read version < N.
    """
    
    def __init__(self, replicas: list, session_id: str):
        self.replicas = replicas
        self.session_id = session_id
        # Track minimum version seen per key
        self.min_versions: dict[str, int] = {}
    
    async def read(self, key: str) -> Any:
        """Read with monotonic guarantee."""
        min_version = self.min_versions.get(key, 0)
        
        # Try replicas until we find one with sufficient version
        for replica in self._shuffled_replicas():
            try:
                result = await replica.read_versioned(key)
                
                if result.version >= min_version:
                    # Update our minimum version
                    self.min_versions[key] = result.version
                    return result.value
                else:
                    # This replica is stale, try another
                    continue
                    
            except Exception:
                continue
        
        # All replicas are stale or unavailable
        raise Exception(f"No replica has version >= {min_version}")
    
    def _shuffled_replicas(self):
        """Shuffle replicas for load balancing."""
        import random
        replicas = self.replicas.copy()
        random.shuffle(replicas)
        return replicas


# Combined: Read-Your-Writes + Monotonic Reads
class SessionConsistencyClient:
    """
    Combines read-your-writes and monotonic reads.
    
    This is what most users expect from a "consistent" system.
    """
    
    def __init__(self, replicas: list, session_id: str):
        self.replicas = replicas
        self.session_id = session_id
        
        # Track writes and read versions
        self.write_versions: dict[str, int] = {}
        self.read_versions: dict[str, int] = {}
    
    async def write(self, key: str, value: Any) -> int:
        """Write and track version."""
        primary = self.replicas[0]
        version = await primary.write(key, value)
        
        self.write_versions[key] = version
        self.read_versions[key] = max(
            self.read_versions.get(key, 0),
            version
        )
        
        return version
    
    async def read(self, key: str) -> Any:
        """Read with session consistency."""
        # Minimum version is max of:
        # - Our last write version
        # - Our last read version
        min_version = max(
            self.write_versions.get(key, 0),
            self.read_versions.get(key, 0)
        )
        
        for replica in self.replicas:
            result = await replica.read_versioned(key)
            if result.version >= min_version:
                self.read_versions[key] = result.version
                return result.value
        
        raise Exception("Consistency requirement not satisfiable")

# Causal Consistency with Vector Clocks

from dataclasses import dataclass, field
from typing import Dict, Any, Optional
import uuid


@dataclass
class VectorClock:
    """
    Vector clock for tracking causal dependencies.
    
    Each node maintains a counter. The vector clock
    is the collection of all node counters.
    """
    clocks: Dict[str, int] = field(default_factory=dict)
    
    def increment(self, node_id: str):
        """Increment this node's clock (on local event)."""
        self.clocks[node_id] = self.clocks.get(node_id, 0) + 1
    
    def merge(self, other: 'VectorClock'):
        """Merge with another vector clock (on receive)."""
        for node_id, count in other.clocks.items():
            self.clocks[node_id] = max(
                self.clocks.get(node_id, 0),
                count
            )
    
    def happens_before(self, other: 'VectorClock') -> bool:
        """Check if self happened before other."""
        # self < other if:
        # - All of self's clocks are <= other's
        # - At least one is strictly <
        all_leq = all(
            self.clocks.get(node, 0) <= other.clocks.get(node, 0)
            for node in set(self.clocks) | set(other.clocks)
        )
        some_lt = any(
            self.clocks.get(node, 0) < other.clocks.get(node, 0)
            for node in set(self.clocks) | set(other.clocks)
        )
        return all_leq and some_lt
    
    def concurrent_with(self, other: 'VectorClock') -> bool:
        """Check if self and other are concurrent."""
        return not self.happens_before(other) and not other.happens_before(self)
    
    def copy(self) -> 'VectorClock':
        return VectorClock(clocks=self.clocks.copy())


@dataclass
class CausalMessage:
    """Message with causal metadata."""
    key: str
    value: Any
    vector_clock: VectorClock
    dependencies: list  # List of message IDs this depends on
    msg_id: str = field(default_factory=lambda: str(uuid.uuid4()))  # Stable across nodes


class CausalConsistencyNode:
    """
    Node implementing causal consistency.
    
    Delays applying updates until their causal
    dependencies are satisfied.
    """
    
    def __init__(self, node_id: str):
        self.node_id = node_id
        self.vector_clock = VectorClock()
        self.data: Dict[str, Any] = {}
        self.pending: list[CausalMessage] = []
        self.delivered: set = set()  # Message IDs we've applied
    
    def local_write(self, key: str, value: Any) -> CausalMessage:
        """Perform a local write operation."""
        # Increment our clock
        self.vector_clock.increment(self.node_id)
        
        # Create message with current vector clock
        msg = CausalMessage(
            key=key,
            value=value,
            vector_clock=self.vector_clock.copy(),
            dependencies=list(self.delivered)
        )
        
        # Apply locally
        self.data[key] = value
        self.delivered.add(msg.msg_id)
        
        return msg  # Broadcast this to other nodes
    
    def receive_message(self, msg: CausalMessage):
        """Receive a message from another node."""
        if self._dependencies_satisfied(msg):
            self._apply_message(msg)
            self._check_pending()
        else:
            # Buffer until dependencies are satisfied
            self.pending.append(msg)
    
    def _dependencies_satisfied(self, msg: CausalMessage) -> bool:
        """Check if all dependencies have been delivered."""
        return all(dep in self.delivered for dep in msg.dependencies)
    
    def _apply_message(self, msg: CausalMessage):
        """Apply a message to local state."""
        self.data[msg.key] = msg.value
        self.vector_clock.merge(msg.vector_clock)
        self.delivered.add(msg.msg_id)
    
    def _check_pending(self):
        """Check if any pending messages can now be applied."""
        made_progress = True
        while made_progress:
            made_progress = False
            for msg in list(self.pending):
                if self._dependencies_satisfied(msg):
                    self._apply_message(msg)
                    self.pending.remove(msg)
                    made_progress = True


# Example usage: Causal social media
class CausalSocialFeed:
    """
    Social feed with causal consistency.
    
    Guarantees: Comments always appear after their parent posts.
    """
    
    def __init__(self, user_id: str, node: CausalConsistencyNode):
        self.user_id = user_id
        self.node = node
    
    def create_post(self, content: str) -> str:
        """Create a new post."""
        post_id = generate_id()
        
        msg = self.node.local_write(
            key=f"post:{post_id}",
            value={
                "id": post_id,
                "author": self.user_id,
                "content": content,
                "type": "post"
            }
        )
        
        # Broadcast msg to other nodes
        return post_id
    
    def comment_on_post(self, post_id: str, content: str) -> str:
        """Comment on an existing post."""
        comment_id = generate_id()
        
        # The comment causally depends on the post
        # (We saw the post, so our vector clock includes it)
        msg = self.node.local_write(
            key=f"comment:{comment_id}",
            value={
                "id": comment_id,
                "parent_post": post_id,
                "author": self.user_id,
                "content": content,
                "type": "comment"
            }
        )
        
        # When other nodes receive this comment,
        # they won't display it until they've seen the parent post
        return comment_id

# Production Consistency-Aware Data Access Layer

import asyncio
from dataclasses import dataclass
from typing import Any, Optional, List, Callable
from enum import Enum
from datetime import datetime
import logging

logger = logging.getLogger(__name__)


class ConsistencyLevel(Enum):
    """Consistency levels for operations."""
    
    EVENTUAL = "eventual"
    # Read from any replica, may be stale
    # Fastest, highest availability
    
    MONOTONIC = "monotonic"
    # Never read older value than previously seen
    # Requires tracking read versions per session
    
    READ_YOUR_WRITES = "read_your_writes"
    # Always see your own writes
    # Requires tracking write versions per session
    
    SESSION = "session"
    # Combination of monotonic + read-your-writes
    # What users typically expect
    
    CAUSAL = "causal"
    # Respects causal ordering
    # Requires vector clocks or dependency tracking
    
    STRONG = "strong"
    # Linearizable, single-copy semantics
    # Slowest, may be unavailable during partitions


@dataclass
class ConsistencyContext:
    """Context for tracking consistency requirements."""
    session_id: str
    write_versions: dict = None
    read_versions: dict = None
    vector_clock: dict = None
    
    def __post_init__(self):
        self.write_versions = self.write_versions or {}
        self.read_versions = self.read_versions or {}
        self.vector_clock = self.vector_clock or {}


class ConsistencyAwareRepository:
    """
    Repository with configurable consistency levels.
    
    Allows different operations to use different consistency
    levels based on their requirements.
    """
    
    def __init__(
        self,
        primary_db,              # Primary database connection
        read_replicas: list,     # Read replica connections
        cache,                   # Cache layer (Redis)
        default_level: ConsistencyLevel = ConsistencyLevel.SESSION
    ):
        self.primary = primary_db
        self.replicas = read_replicas
        self.cache = cache
        self.default_level = default_level
    
    async def read(
        self,
        key: str,
        consistency: ConsistencyLevel = None,
        context: ConsistencyContext = None
    ) -> Optional[Any]:
        """
        Read with specified consistency level.
        """
        level = consistency or self.default_level
        context = context or ConsistencyContext(session_id="default")
        
        if level == ConsistencyLevel.EVENTUAL:
            return await self._read_eventual(key)
        
        elif level == ConsistencyLevel.MONOTONIC:
            return await self._read_monotonic(key, context)
        
        elif level == ConsistencyLevel.READ_YOUR_WRITES:
            return await self._read_your_writes(key, context)
        
        elif level == ConsistencyLevel.SESSION:
            return await self._read_session(key, context)
        
        elif level == ConsistencyLevel.STRONG:
            return await self._read_strong(key)
        
        else:
            raise ValueError(f"Unknown consistency level: {level}")
    
    async def write(
        self,
        key: str,
        value: Any,
        consistency: ConsistencyLevel = None,
        context: ConsistencyContext = None
    ) -> int:
        """
        Write with specified consistency level.
        Returns version number of the write.
        """
        level = consistency or self.default_level
        context = context or ConsistencyContext(session_id="default")
        
        if level == ConsistencyLevel.STRONG:
            version = await self._write_strong(key, value)
        else:
            version = await self._write_async(key, value)
        
        # Update context for read-your-writes
        context.write_versions[key] = version
        context.read_versions[key] = max(
            context.read_versions.get(key, 0),
            version
        )
        
        return version
    
    # =========================================================================
    # Eventual Consistency
    # =========================================================================
    
    async def _read_eventual(self, key: str) -> Optional[Any]:
        """Read from cache or any replica. May be stale."""
        # Try cache first
        cached = await self.cache.get(key)
        if cached is not None:
            return cached
        
        # Try replicas (load balanced)
        import random
        replicas = self.replicas.copy()
        random.shuffle(replicas)
        
        for replica in replicas:
            try:
                result = await replica.fetch_one(
                    "SELECT value, version FROM data WHERE key = $1",
                    key
                )
                if result:
                    # Cache for future reads
                    await self.cache.set(key, result['value'], ttl=60)
                    return result['value']
            except Exception as e:
                logger.warning(f"Replica read failed: {e}")
                continue
        
        # Fall back to primary
        return await self._read_from_primary(key)
    
    # =========================================================================
    # Monotonic Reads
    # =========================================================================
    
    async def _read_monotonic(
        self,
        key: str,
        context: ConsistencyContext
    ) -> Optional[Any]:
        """Read ensuring we never go backwards."""
        min_version = context.read_versions.get(key, 0)
        
        # Try replicas that have sufficient version
        for replica in self.replicas:
            try:
                result = await replica.fetch_one(
                    """
                    SELECT value, version FROM data 
                    WHERE key = $1 AND version >= $2
                    """,
                    key, min_version
                )
                if result:
                    context.read_versions[key] = result['version']
                    return result['value']
            except Exception:
                continue
        
        # Fall back to primary (always has latest)
        result = await self._read_from_primary_versioned(key)
        if result and result['version'] >= min_version:
            context.read_versions[key] = result['version']
            return result['value']
        
        return None
    
    # =========================================================================
    # Read-Your-Writes
    # =========================================================================
    
    async def _read_your_writes(
        self,
        key: str,
        context: ConsistencyContext
    ) -> Optional[Any]:
        """Read ensuring we see our own writes."""
        write_version = context.write_versions.get(key)
        
        if write_version is None:
            # We haven't written this key, any replica is fine
            return await self._read_eventual(key)
        
        # We wrote this key, ensure we see our write
        return await self._read_with_min_version(key, write_version)
    
    async def _read_with_min_version(
        self,
        key: str,
        min_version: int,
        context: ConsistencyContext = None
    ) -> Optional[Any]:
        """Read from replica with at least min_version."""
        # Try replicas
        for replica in self.replicas:
            result = await replica.fetch_one(
                """
                SELECT value, version FROM data 
                WHERE key = $1 AND version >= $2
                """,
                key, min_version
            )
            if result:
                break
        else:
            # Replicas don't have our version yet, use primary
            result = await self._read_from_primary_versioned(key)
        
        if result and context:
            # Record the version actually seen (for monotonic reads)
            context.read_versions[key] = result['version']
        return result['value'] if result else None
    
    # =========================================================================
    # Session Consistency (Monotonic + Read-Your-Writes)
    # =========================================================================
    
    async def _read_session(
        self,
        key: str,
        context: ConsistencyContext
    ) -> Optional[Any]:
        """Read with session consistency guarantees."""
        # Min version is max of write version and read version
        min_version = max(
            context.write_versions.get(key, 0),
            context.read_versions.get(key, 0)
        )
        
        if min_version == 0:
            return await self._read_eventual(key)
        
        # Records the version actually read, so later reads never go backwards
        return await self._read_with_min_version(key, min_version, context)
    
    # =========================================================================
    # Strong Consistency
    # =========================================================================
    
    async def _read_strong(self, key: str) -> Optional[Any]:
        """Read from primary only. Linearizable."""
        return await self._read_from_primary(key)
    
    async def _write_strong(self, key: str, value: Any) -> int:
        """Write with synchronous replication."""
        async with self.primary.transaction():
            # Write to primary
            result = await self.primary.fetch_one(
                """
                INSERT INTO data (key, value, version)
                VALUES ($1, $2, 1)
                ON CONFLICT (key) DO UPDATE 
                SET value = $2, version = data.version + 1
                RETURNING version
                """,
                key, value
            )
            version = result['version']
            
            # Synchronously replicate to all replicas
            # (This is slow but guarantees strong consistency)
            await self._sync_replicate(key, value, version)
            
            # Invalidate cache
            await self.cache.delete(key)
            
            return version
    
    async def _write_async(self, key: str, value: Any) -> int:
        """Write to primary with async replication."""
        result = await self.primary.fetch_one(
            """
            INSERT INTO data (key, value, version)
            VALUES ($1, $2, 1)
            ON CONFLICT (key) DO UPDATE 
            SET value = $2, version = data.version + 1
            RETURNING version
            """,
            key, value
        )
        version = result['version']
        
        # Async replication (don't wait)
        asyncio.create_task(self._async_replicate(key, value, version))
        
        # Invalidate cache
        await self.cache.delete(key)
        
        return version
    
    # =========================================================================
    # Helper Methods
    # =========================================================================
    
    async def _read_from_primary(self, key: str) -> Optional[Any]:
        """Read directly from primary."""
        result = await self.primary.fetch_one(
            "SELECT value FROM data WHERE key = $1",
            key
        )
        return result['value'] if result else None
    
    async def _read_from_primary_versioned(self, key: str) -> Optional[dict]:
        """Read with version from primary."""
        return await self.primary.fetch_one(
            "SELECT value, version FROM data WHERE key = $1",
            key
        )
    
    async def _sync_replicate(self, key: str, value: Any, version: int):
        """Synchronously replicate to all replicas."""
        await asyncio.gather(*[
            replica.execute(
                """
                INSERT INTO data (key, value, version) VALUES ($1, $2, $3)
                ON CONFLICT (key) DO UPDATE SET value = $2, version = $3
                WHERE data.version < $3
                """,
                key, value, version
            )
            for replica in self.replicas
        ])
    
    async def _async_replicate(self, key: str, value: Any, version: int):
        """Asynchronously replicate to replicas."""
        for replica in self.replicas:
            try:
                await replica.execute(
                    """
                    INSERT INTO data (key, value, version) VALUES ($1, $2, $3)
                    ON CONFLICT (key) DO UPDATE SET value = $2, version = $3
                    WHERE data.version < $3
                    """,
                    key, value, version
                )
            except Exception as e:
                logger.warning(f"Async replication failed: {e}")

# Inventory with Appropriate Consistency Levels

class InventoryService:
    """
    Inventory service with consistency-aware operations.
    
    Different operations have different consistency requirements:
    - Display: Eventual consistency (from cache)
    - Reservation: Strong consistency (prevent oversell)
    - Release: Strong consistency (prevent under-count)
    """
    
    def __init__(
        self,
        repository: ConsistencyAwareRepository,
        cache
    ):
        self.repo = repository
        self.cache = cache
    
    async def get_display_count(self, product_id: str) -> int:
        """
        Get inventory count for DISPLAY purposes.
        
        Eventual consistency is fine - showing stale count
        on a product page is acceptable.
        """
        # Try cache first (eventual consistency)
        cache_key = f"inventory:{product_id}"
        cached = await self.cache.get(cache_key)
        
        if cached is not None:
            return cached
        
        # Read from replica (eventual consistency)
        count = await self.repo.read(
            key=f"inventory:{product_id}",
            consistency=ConsistencyLevel.EVENTUAL
        )
        
        # Cache for future reads
        if count is not None:
            await self.cache.set(cache_key, count, ttl=30)
        
        return count or 0
    
    async def reserve(
        self,
        product_id: str,
        quantity: int,
        reservation_id: str
    ) -> bool:
        """
        Reserve inventory for checkout.
        
        STRONG consistency required - we must prevent overselling.
        Uses pessimistic locking (SELECT ... FOR UPDATE) on the primary.
        """
        async with self.repo.primary.transaction(isolation="SERIALIZABLE"):
            # Read current inventory with lock
            result = await self.repo.primary.fetch_one(
                """
                SELECT quantity, version FROM inventory 
                WHERE product_id = $1 FOR UPDATE
                """,
                product_id
            )
            
            if not result or result['quantity'] < quantity:
                return False
            
            # Decrement inventory
            await self.repo.primary.execute(
                """
                UPDATE inventory 
                SET quantity = quantity - $2, version = version + 1
                WHERE product_id = $1
                """,
                product_id, quantity
            )
            
            # Record reservation
            await self.repo.primary.execute(
                """
                INSERT INTO reservations (id, product_id, quantity, status, created_at)
                VALUES ($1, $2, $3, 'active', NOW())
                """,
                reservation_id, product_id, quantity
            )
            
            # Invalidate cache
            await self.cache.delete(f"inventory:{product_id}")
            
            return True
    
    async def release(self, reservation_id: str) -> bool:
        """
        Release a reservation (order cancelled).
        
        STRONG consistency - must accurately restore inventory.
        """
        async with self.repo.primary.transaction():
            # Get reservation
            reservation = await self.repo.primary.fetch_one(
                """
                SELECT product_id, quantity, status FROM reservations
                WHERE id = $1 FOR UPDATE
                """,
                reservation_id
            )
            
            if not reservation or reservation['status'] != 'active':
                return False
            
            # Restore inventory
            await self.repo.primary.execute(
                """
                UPDATE inventory 
                SET quantity = quantity + $2, version = version + 1
                WHERE product_id = $1
                """,
                reservation['product_id'], reservation['quantity']
            )
            
            # Mark reservation as released
            await self.repo.primary.execute(
                """
                UPDATE reservations SET status = 'released' WHERE id = $1
                """,
                reservation_id
            )
            
            # Invalidate cache
            await self.cache.delete(f"inventory:{reservation['product_id']}")
            
            return True
    
    async def confirm(self, reservation_id: str) -> bool:
        """
        Confirm a reservation (order completed).
        
        Changes reservation status, inventory already decremented.
        """
        result = await self.repo.primary.execute(
            """
            UPDATE reservations SET status = 'confirmed'
            WHERE id = $1 AND status = 'active'
            """,
            reservation_id
        )
        return result.rowcount > 0

# User Profile with Read-Your-Writes

class ProfileService:
    """
    User profile service with session consistency.
    
    Users expect to see their own updates immediately.
    Other users' updates can be eventually consistent.
    """
    
    def __init__(self, repository: ConsistencyAwareRepository):
        self.repo = repository
    
    async def get_profile(
        self,
        user_id: str,
        viewer_id: str,
        context: ConsistencyContext
    ) -> dict:
        """
        Get a user's profile.
        
        If viewing your own profile: Read-your-writes
        If viewing someone else's: Eventual consistency
        """
        if user_id == viewer_id:
            # Viewing own profile - see your updates
            consistency = ConsistencyLevel.READ_YOUR_WRITES
        else:
            # Viewing someone else - eventual is fine
            consistency = ConsistencyLevel.EVENTUAL
        
        return await self.repo.read(
            key=f"profile:{user_id}",
            consistency=consistency,
            context=context
        )
    
    async def update_profile(
        self,
        user_id: str,
        updates: dict,
        context: ConsistencyContext
    ) -> dict:
        """Update user's profile."""
        # Merge with existing profile
        current = await self.repo.read(
            key=f"profile:{user_id}",
            consistency=ConsistencyLevel.STRONG,  # Get latest for merge
            context=context
        )
        
        updated = {**(current or {}), **updates}
        
        await self.repo.write(
            key=f"profile:{user_id}",
            value=updated,
            context=context  # Track for read-your-writes
        )
        
        return updated

# Quorum-Based Consistency Implementation

from dataclasses import dataclass
from typing import List, Optional, Any, Tuple
import asyncio
import time


@dataclass
class QuorumConfig:
    """Configuration for quorum-based consistency."""
    n: int  # Total replicas
    w: int  # Write quorum
    r: int  # Read quorum
    
    def is_strong(self) -> bool:
        """Check if config provides strong consistency."""
        return self.w + self.r > self.n
    
    def __post_init__(self):
        if self.w > self.n or self.r > self.n:
            raise ValueError("Quorum cannot exceed replica count")


@dataclass
class VersionedValue:
    """Value with version for conflict detection."""
    value: Any
    version: int
    timestamp: float


class QuorumStore:
    """
    Distributed store with quorum-based consistency.
    
    Provides tunable consistency through W and R parameters.
    """
    
    def __init__(self, replicas: List, config: QuorumConfig):
        self.replicas = replicas
        self.config = config
    
    async def write(self, key: str, value: Any) -> bool:
        """
        Write to W replicas (quorum write).
        
        Returns True if quorum was achieved.
        """
        timestamp = time.time()
        
        # Get current max version
        current = await self._read_from_quorum(key)
        new_version = (current.version + 1) if current else 1
        
        versioned = VersionedValue(
            value=value,
            version=new_version,
            timestamp=timestamp
        )
        
        # Write to all replicas concurrently
        tasks = [
            self._write_to_replica(replica, key, versioned)
            for replica in self.replicas
        ]
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Count successful writes
        successes = sum(1 for r in results if r is True)
        
        if successes >= self.config.w:
            return True
        else:
            # Quorum not achieved - write failed
            # Could attempt rollback here
            return False
    
    async def read(self, key: str) -> Optional[Any]:
        """
        Read from R replicas (quorum read).
        
        Returns the value with highest version.
        """
        result = await self._read_from_quorum(key)
        return result.value if result else None
    
    async def _read_from_quorum(self, key: str) -> Optional[VersionedValue]:
        """Read from R replicas and return highest version."""
        tasks = [
            self._read_from_replica(replica, key)
            for replica in self.replicas
        ]
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Filter successful reads (keep each value paired with its replica)
        pairs = [
            (replica, r) for replica, r in zip(self.replicas, results)
            if isinstance(r, VersionedValue)
        ]
        values = [v for _, v in pairs]
        
        if len(values) < self.config.r:
            raise Exception(f"Read quorum not achieved: {len(values)}/{self.config.r}")
        
        # Return highest version (read repair could update stale replicas)
        if not values:
            return None
        
        latest = max(values, key=lambda v: (v.version, v.timestamp))
        
        # Optional: Read repair - update stale replicas
        await self._read_repair(key, latest, pairs)
        
        return latest
    
    async def _read_repair(
        self,
        key: str,
        latest: VersionedValue,
        pairs: List[Tuple[Any, VersionedValue]]
    ):
        """Update replicas that have stale values."""
        stale_replicas = []
        
        for replica, value in pairs:
            if value.version < latest.version:
                stale_replicas.append(replica)
        
        if stale_replicas:
            # Repair stale replicas asynchronously
            asyncio.create_task(
                self._repair_replicas(key, latest, stale_replicas)
            )
    
    async def _repair_replicas(
        self,
        key: str,
        value: VersionedValue,
        replicas: List
    ):
        """Repair stale replicas with latest value."""
        tasks = [
            self._write_to_replica(replica, key, value)
            for replica in replicas
        ]
        await asyncio.gather(*tasks, return_exceptions=True)
    
    async def _write_to_replica(
        self,
        replica,
        key: str,
        value: VersionedValue
    ) -> bool:
        """Write to a single replica."""
        try:
            await replica.set(key, value)
            return True
        except Exception as e:
            logger.warning(f"Write to replica failed: {e}")
            return False
    
    async def _read_from_replica(
        self,
        replica,
        key: str
    ) -> Optional[VersionedValue]:
        """Read from a single replica."""
        try:
            return await replica.get(key)
        except Exception as e:
            logger.warning(f"Read from replica failed: {e}")
            raise


# Usage examples
async def demo_quorum_consistency():
    # Strong consistency: W=2, R=2 with N=3
    strong_config = QuorumConfig(n=3, w=2, r=2)
    strong_store = QuorumStore(replicas, strong_config)
    
    # Write requires 2/3 replicas
    await strong_store.write("key", "value")
    
    # Read requires 2/3 replicas - guaranteed to see latest write
    value = await strong_store.read("key")
    
    
    # Eventual consistency: W=1, R=1 with N=3
    eventual_config = QuorumConfig(n=3, w=1, r=1)
    eventual_store = QuorumStore(replicas, eventual_config)
    
    # Fast writes and reads, but may return stale data
    await eventual_store.write("key", "value")
    value = await eventual_store.read("key")  # Might be stale!

# Sloppy Quorums for High Availability

class SloppyQuorumStore(QuorumStore):
    """
    Quorum store with sloppy quorums and hinted handoff.
    
    When a replica is down, writes go to a "hint" node instead.
    When the original replica recovers, hints are handed off.
    
    This improves availability at the cost of consistency guarantees.
    """
    
    def __init__(self, replicas: List, config: QuorumConfig, all_nodes: List):
        super().__init__(replicas, config)
        self.all_nodes = all_nodes  # Including non-replica nodes
        self.hints: dict = {}  # node_id -> list of hints
    
    async def write_sloppy(self, key: str, value: Any) -> bool:
        """
        Write with sloppy quorum.
        
        If a replica is down, write to any available node as a hint.
        """
        timestamp = time.time()
        current = await self._read_from_quorum(key)
        new_version = (current.version + 1) if current else 1
        
        versioned = VersionedValue(
            value=value,
            version=new_version,
            timestamp=timestamp
        )
        
        successes = 0
        hints_stored = []
        
        # Try preferred replicas first
        for replica in self.replicas:
            if await self._write_to_replica(replica, key, versioned):
                successes += 1
            else:
                # Replica is down, find a substitute
                substitute = await self._find_substitute(replica)
                if substitute:
                    await self._store_hint(substitute, replica.id, key, versioned)
                    successes += 1
                    hints_stored.append((substitute, replica.id))
        
        return successes >= self.config.w
    
    async def _find_substitute(self, failed_replica) -> Optional:
        """Find an available node to store hint."""
        for node in self.all_nodes:
            if node.id != failed_replica.id and await node.is_healthy():
                return node
        return None
    
    async def _store_hint(
        self,
        node,
        target_replica_id: str,
        key: str,
        value: VersionedValue
    ):
        """Store a hinted value for later handoff."""
        hint = {
            "target": target_replica_id,
            "key": key,
            "value": value
        }
        
        # Store hint on the substitute node
        await node.store_hint(hint)
    
    async def handoff_hints(self, recovered_replica):
        """
        Hand off hints to a recovered replica.
        
        Called when a replica comes back online.
        """
        for node in self.all_nodes:
            hints = await node.get_hints_for(recovered_replica.id)
            
            for hint in hints:
                try:
                    await self._write_to_replica(
                        recovered_replica,
                        hint["key"],
                        hint["value"]
                    )
                    await node.delete_hint(hint)
                except Exception as e:
                    logger.warning(f"Hint handoff failed: {e}")

# Isolation Level Selection by Use Case

class TransactionManager:
    """
    Manager for database transactions with appropriate isolation.
    """
    
    def __init__(self, db_pool):
        self.db = db_pool
    
    async def browse_products(self, category_id: str) -> List[dict]:
        """
        Browsing products - READ COMMITTED is fine.
        
        We don't need repeatable reads for browsing.
        Seeing slightly different results on refresh is OK.
        """
        async with self.db.transaction(isolation="READ COMMITTED"):
            return await self.db.fetch(
                "SELECT * FROM products WHERE category_id = $1",
                category_id
            )
    
    async def generate_report(self, start_date: str, end_date: str) -> dict:
        """
        Generating a report - REPEATABLE READ.
        
        Report should be internally consistent.
        Same query should return same results throughout.
        """
        async with self.db.transaction(isolation="REPEATABLE READ"):
            total_sales = await self.db.fetch_val(
                "SELECT SUM(amount) FROM orders WHERE date BETWEEN $1 AND $2",
                start_date, end_date
            )
            
            top_products = await self.db.fetch(
                """
                SELECT product_id, SUM(quantity) as total
                FROM order_items oi
                JOIN orders o ON oi.order_id = o.id
                WHERE o.date BETWEEN $1 AND $2
                GROUP BY product_id
                ORDER BY total DESC
                LIMIT 10
                """,
                start_date, end_date
            )
            
            # Both queries see the same snapshot
            return {"total": total_sales, "top_products": top_products}
    
    async def reserve_inventory(
        self,
        product_id: str,
        quantity: int
    ) -> bool:
        """
        Reserving inventory - SERIALIZABLE.
        
        Must prevent race conditions that cause overselling.
        """
        async with self.db.transaction(isolation="SERIALIZABLE"):
            # Consistent snapshot; a conflicting concurrent transaction is aborted
            # with a serialization failure (SQLSTATE 40001), so callers must retry
            current = await self.db.fetch_one(
                "SELECT quantity FROM inventory WHERE product_id = $1",
                product_id
            )
            
            if current['quantity'] < quantity:
                return False
            
            await self.db.execute(
                "UPDATE inventory SET quantity = quantity - $1 WHERE product_id = $2",
                quantity, product_id
            )
            
            return True
    
    async def transfer_funds(
        self,
        from_account: str,
        to_account: str,
        amount: float
    ) -> bool:
        """
        Money transfer - SERIALIZABLE with explicit locks.
        
        Double protection: isolation level + explicit locks.
        """
        async with self.db.transaction(isolation="SERIALIZABLE"):
            # Lock both accounts in consistent order to prevent deadlock
            accounts = sorted([from_account, to_account])
            
            balances = await self.db.fetch(
                """
                SELECT account_id, balance FROM accounts 
                WHERE account_id = ANY($1)
                ORDER BY account_id
                FOR UPDATE
                """,
                accounts
            )
            
            balance_map = {row['account_id']: row['balance'] for row in balances}
            
            if balance_map[from_account] < amount:
                raise InsufficientFundsError()
            
            await self.db.execute(
                "UPDATE accounts SET balance = balance - $1 WHERE account_id = $2",
                amount, from_account
            )
            
            await self.db.execute(
                "UPDATE accounts SET balance = balance + $1 WHERE account_id = $2",
                amount, to_account
            )
            
            return True

# Optimistic vs Pessimistic Locking Comparison

class LockingPatterns:
    """
    Comparison of locking strategies for consistency.
    """
    
    # =========================================================================
    # Pessimistic Locking: Lock first, then read/write
    # =========================================================================
    
    async def pessimistic_update(self, product_id: str, new_price: float):
        """
        Pessimistic locking with SELECT FOR UPDATE.
        
        Use when:
        - High contention expected
        - Short transaction duration
        - Can tolerate lock waiting
        """
        async with self.db.transaction():
            # Lock the row immediately
            product = await self.db.fetch_one(
                """
                SELECT * FROM products 
                WHERE id = $1 
                FOR UPDATE  -- Acquires exclusive lock
                """,
                product_id
            )
            
            if not product:
                raise NotFoundError()
            
            # Other transactions wait here until we commit
            await self.db.execute(
                "UPDATE products SET price = $1 WHERE id = $2",
                new_price, product_id
            )
    
    async def pessimistic_with_nowait(self, product_id: str, new_price: float):
        """
        Pessimistic locking with NOWAIT - fail fast if locked.
        
        Use when:
        - Can't afford to wait
        - Will retry or fail gracefully
        """
        async with self.db.transaction():
            try:
                product = await self.db.fetch_one(
                    """
                    SELECT * FROM products 
                    WHERE id = $1 
                    FOR UPDATE NOWAIT  -- Fails immediately if locked
                    """,
                    product_id
                )
            except LockNotAvailableError:
                raise ConflictError("Product is being modified")
            
            await self.db.execute(
                "UPDATE products SET price = $1 WHERE id = $2",
                new_price, product_id
            )
    
    # =========================================================================
    # Optimistic Locking: Read version, check at write time
    # =========================================================================
    
    async def optimistic_update(
        self,
        product_id: str,
        new_price: float,
        expected_version: int
    ):
        """
        Optimistic locking with version check.
        
        Use when:
        - Low contention expected
        - Long-running operations (user editing form)
        - Can retry on conflict
        """
        result = await self.db.execute(
            """
            UPDATE products 
            SET price = $1, version = version + 1
            WHERE id = $2 AND version = $3
            """,
            new_price, product_id, expected_version
        )
        
        if result.rowcount == 0:
            # Either not found or version mismatch
            current = await self.db.fetch_one(
                "SELECT version FROM products WHERE id = $1",
                product_id
            )
            
            if current is None:
                raise NotFoundError()
            else:
                raise OptimisticLockError(
                    f"Version mismatch: expected {expected_version}, "
                    f"got {current['version']}"
                )
    
    async def optimistic_with_retry(
        self,
        product_id: str,
        update_fn: Callable,
        max_retries: int = 3
    ):
        """
        Optimistic locking with automatic retry.
        
        Retries on conflict with exponential backoff.
        """
        for attempt in range(max_retries):
            # Read current state
            product = await self.db.fetch_one(
                "SELECT * FROM products WHERE id = $1",
                product_id
            )
            
            if not product:
                raise NotFoundError()
            
            # Apply update function
            updated = update_fn(product)
            
            # Try to save with version check
            result = await self.db.execute(
                """
                UPDATE products 
                SET price = $1, name = $2, version = version + 1
                WHERE id = $3 AND version = $4
                """,
                updated['price'], updated['name'],
                product_id, product['version']
            )
            
            if result.rowcount > 0:
                return updated  # Success!
            
            # Conflict - retry with backoff
            await asyncio.sleep(0.1 * (2 ** attempt))
        
        raise OptimisticLockError(f"Failed after {max_retries} retries")