状态管理：分布式状态与一致性

{
  "task_id": "task_001",
  "type": "code_generation",
  "status": "running",
  "progress": 0.65,
  "created_at": "2026-05-25T10:00:00Z",
  "metadata": {
    "language": "python",
    "framework": "fastapi",
    "requirements": ["生成RESTful API", "包含JWT认证"]
  },
  "dependencies": [
    {"task_id": "task_000", "type": "requirement_analysis"},
    {"task_id": "task_003", "type": "database_design"}
  ],
  "artifacts": [
    {"path": "/src/api/main.py", "lines": 234, "sha": "abc123"},
    {"path": "/src/api/routes.py", "lines": 156, "sha": "def456"}
  ],
  "checkpoint": {
    "last_saved": "2026-05-25T10:15:00Z",
    "snapshot_id": "snap_xyz789"
  }
}

# 状态模型配置
STATE_MODELS = {
    # 关系型：用户、会话、权限
    "relational": ["User", "Session", "Permission", "Role"],
    
    # 文档型：任务、Agent记忆、工作区
    "document": ["Task", "AgentMemory", "Workspace", "Checkpoint"],
    
    # KV型：缓存、锁、计数器
    "keyvalue": ["Cache", "Lock", "Counter", "RateLimit"],
    
    # 消息型：操作日志、审计
    "log": ["OperationLog", "AuditLog", "Metrics"]
}

# 内存状态管理器示例
class InMemoryStateManager:
    """轻量级内存状态管理器，适用于单进程场景"""
    
    def __init__(self):
        self._state: Dict[str, Any] = {}
        self._locks: Dict[str, asyncio.Lock] = {}
        self._version: Dict[str, int] = {}
    
    async def get(self, key: str) -> Optional[Any]:
        """读取状态"""
        return self._state.get(key)
    
    async def set(self, key: str, value: Any, version: int = None) -> bool:
        """写入状态，支持版本检查"""
        if version is not None:
            if self._version.get(key, 0) != version:
                return False  # 版本冲突
        
        self._state[key] = value
        self._version[key] = version + 1 if version is not None else 1
        return True
    
    async def compare_and_set(self, key: str, expected: Any, new_value: Any) -> bool:
        """CAS操作：仅当当前值等于expected时设置新值"""
        current = self._state.get(key)
        if current != expected:
            return False
        self._state[key] = new_value
        return True
    
    async def acquire_lock(self, key: str, timeout: float = 10.0) -> bool:
        """获取分布式锁（单机模拟）"""
        if key not in self._locks:
            self._locks[key] = asyncio.Lock()
        try:
            return await asyncio.wait_for(
                self._locks[key].acquire(),
                timeout=timeout
            )
        except asyncio.TimeoutError:
            return False
    
    async def release_lock(self, key: str) -> None:
        """释放锁"""
        if key in self._locks and self._locks[key].locked():
            self._locks[key].release()

# WAL持久化示例
import os
import json
import struct
from typing import Any, Optional
from dataclasses import dataclass
from enum import Enum

class LogOp(Enum):
    SET = 0x01
    DELETE = 0x02
    SNAPSHOT = 0x03

@dataclass
class WALEntry:
    """WAL日志条目"""
    term: int
    operation: LogOp
    key: str
    value: Optional[bytes]
    timestamp: int
    
    def serialize(self) -> bytes:
        """序列化WAL条目"""
        header = struct.pack(
            '>BQiI',  # 大端序
            self.term,
            self.operation.value,
            len(self.key),
            self.timestamp
        )
        key_bytes = self.key.encode('utf-8')
        value_bytes = self.value or b''
        return header + key_bytes + value_bytes
    
    @classmethod
    def deserialize(cls, data: bytes) -> 'WALEntry':
        """反序列化WAL条目"""
        term, op_val, key_len, timestamp = struct.unpack('>BQiI', data[:18])
        key = data[18:18+key_len].decode('utf-8')
        value = data[18+key_len:] if len(data) > 18+key_len else None
        return cls(term, LogOp(op_val), key, value, timestamp)

class WALPersistence:
    """Write-Ahead Log持久化引擎"""
    
    def __init__(self, log_dir: str):
        self.log_dir = log_dir
        self.current_log = os.path.join(log_dir, "wal.log")
        self.snapshot_dir = os.path.join(log_dir, "snapshots")
        self._ensure_dirs()
        
        # 日志序列号
        self._log_index = 0
        self._last_snapshot_index = 0
    
    def _ensure_dirs(self):
        os.makedirs(self.log_dir, exist_ok=True)
        os.makedirs(self.snapshot_dir, exist_ok=True)
    
    def write(self, entry: WALEntry) -> int:
        """追加写日志"""
        with open(self.current_log, 'ab') as f:
            offset = f.tell()
            f.write(entry.serialize())
            self._log_index += 1
            return offset
    
    def read_from(self, start_index: int) -> list[WALEntry]:
        """从指定索引读取日志"""
        entries = []
        with open(self.current_log, 'rb') as f:
            while True:
                header = f.read(18)
                if not header:
                    break
                term, op_val, key_len, timestamp = struct.unpack('>BQiI', header)
                key_data = f.read(key_len)
                value_len = (os.path.getsize(self.current_log) - f.tell())
                value = f.read(value_len) if value_len > 0 else None
                
                entry = WALEntry(term, LogOp(op_val), key_data.decode('utf-8'), value, timestamp)
                if self._log_index >= start_index:
                    entries.append(entry)
        return entries
    
    def create_snapshot(self, state: dict, index: int) -> str:
        """创建快照"""
        snapshot_path = os.path.join(self.snapshot_dir, f"snapshot_{index}.json")
        with open(snapshot_path, 'w') as f:
            json.dump(state, f)
        self._last_snapshot_index = index
        return snapshot_path
    
    def load_latest_snapshot(self) -> Optional[dict]:
        """加载最新快照"""
        snapshots = sorted(os.listdir(self.snapshot_dir))
        if not snapshots:
            return None
        latest = snapshots[-1]
        with open(os.path.join(self.snapshot_dir, latest), 'r') as f:
            return json.load(f)

# Redis数据结构使用映射
REDIS_DATA_STRUCTURES = {
    # String: 简单值、计数器、分布式锁
    "string": {
        "session:token:{user_id}": "JWT token值",
        "task:counter": "任务序号计数器",
        "lock:task:{task_id}": "任务分布式锁",
    },
    
    # Hash: 对象、实体
    "hash": {
        "session:{session_id}": {
            "user_id": "用户ID",
            "created_at": "创建时间",
            "last_active": "最后活跃时间",
            "state": "会话状态",
        },
        "task:{task_id}": {
            "type": "任务类型",
            "status": "pending/running/completed",
            "progress": "0.0-1.0",
        },
    },
    
    # List: 队列、消息列表
    "list": {
        "queue:pending": "待执行任务队列",
        "queue:completed": "已完成任务列表",
        "log:session:{session_id}": "会话操作日志",
    },
    
    # Set: 标签、去重
    "set": {
        "tags:task:{task_id}": ["代码生成", "Python", "API"],
        "users:active": "当前活跃用户集合",
    },
    
    # Sorted Set: 排行榜、优先级队列
    "zset": {
        "queue:priority": "优先级任务队列(score=优先级)",
        "leaderboard:agents": "Agent性能排行榜",
    },
    
    # Stream: 事件流、日志
    "stream": {
        "events:task": "任务状态变更事件流",
        "events:agent": "Agent生命周期事件流",
    },
    
    # Bitmap: 状态位图、在线状态
    "bitmap": {
        "online:users": "用户在线状态位图",
        "feature:flags": "特性开关位图",
    },
    
    # HyperLogLog: 统计去重
    "hll": {
        "uv:daily": "日活用户统计",
        "uv:monthly": "月活用户统计",
    },
}

import redis
from typing import List, Tuple, Any
import json

class RedisStateManager:
    """基于Redis的高级状态管理器"""
    
    def __init__(self, redis_url: str):
        self.pool = redis.ConnectionPool.from_url(redis_url)
        self.client = redis.Redis(connection_pool=self.pool)
    
    # ==================== 批量操作优化 ====================
    
    def batch_get_tasks(self, task_ids: List[str]) -> dict[str, dict]:
        """批量获取任务状态 - 使用Pipeline"""
        pipe = self.client.pipeline()
        for task_id in task_ids:
            pipe.hgetall(f"task:{task_id}")
        
        results = pipe.execute()
        return {
            task_id: result 
            for task_id, result in zip(task_ids, results) 
            if result  # 过滤空结果
        }
    
    def batch_update_progress(self, updates: List[Tuple[str, float]]) -> int:
        """批量更新任务进度 - 使用Pipeline"""
        pipe = self.client.pipeline()
        for task_id, progress in updates:
            pipe.hset(f"task:{task_id}", "progress", progress)
            pipe.hset(f"task:{task_id}", "updated_at", int(time.time()))
        results = pipe.execute()
        return sum(1 for r in results if r)  # 返回成功更新的数量
    
    # ==================== Lua脚本原子操作 ====================
    
    UPDATE_TASK_STATUS_LUA = """
    local task_key = KEYS[1]
    local new_status = ARGV[1]
    local expected_status = ARGV[2]
    local progress = ARGV[3]
    local version = ARGV[4]
    
    -- 版本检查（乐观锁）
    local current_version = redis.call('HGET', task_key, 'version')
    if current_version and tonumber(current_version) ~= tonumber(version) then
        return {err = 'VERSION_CONFLICT', current_version}
    end
    
    -- 状态转换验证
    local current_status = redis.call('HGET', task_key, 'status')
    if current_status ~= expected_status then
        return {err = 'STATUS_CONFLICT', current_status}
    end
    
    -- 更新状态
    redis.call('HSET', task_key, 'status', new_status)
    redis.call('HSET', task_key, 'progress', progress)
    redis.call('HSET', task_key, 'version', version + 1)
    redis.call('HSET', task_key, 'updated_at', ARGV[5])
    
    return {ok = 'SUCCESS', version + 1}
    """
    
    def update_task_status_atomic(
        self, 
        task_id: str, 
        new_status: str, 
        expected_status: str,
        progress: float,
        version: int
    ) -> dict:
        """原子性地更新任务状态（支持乐观锁）"""
        script = self.client.register_script(self.UPDATE_TASK_STATUS_LUA)
        result = script(
            keys=[f"task:{task_id}"],
            args=[new_status, expected_status, progress, version, int(time.time())]
        )
        
        if isinstance(result, list) and len(result) == 2:
            if result[0] == b'err':
                return {"success": False, "error": result[1].decode()}
            elif result[0] == b'ok':
                return {"success": True, "new_version": result[1]}
        
        return {"success": False, "error": "UNKNOWN"}
    
    # ==================== 分布式锁实现 ====================
    
    ACQUIRE_LOCK_LUA = """
    local lock_key = KEYS[1]
    local token = ARGV[1]
    local ttl_ms = ARGV[2]
    
    -- SET NX EX 原子操作
    local result = redis.call('SET', lock_key, token, 'NX', 'PX', ttl_ms)
    if result then
        return 1
    else
        return 0
    end
    """
    
    RELEASE_LOCK_LUA = """
    local lock_key = KEYS[1]
    local token = ARGV[1]
    
    -- Lua脚本保证检查和删除的原子性
    local current_token = redis.call('GET', lock_key)
    if current_token == token then
        redis.call('DEL', lock_key)
        return 1
    else
        return 0
    end
    """
    
    def acquire_lock(
        self, 
        resource: str, 
        token: str, 
        ttl_ms: int = 30000
    ) -> bool:
        """获取分布式锁"""
        script = self.client.register_script(self.ACQUIRE_LOCK_LUA)
        return bool(script(keys=[f"lock:{resource}"], args=[token, ttl_ms]))
    
    def release_lock(self, resource: str, token: str) -> bool:
        """释放分布式锁"""
        script = self.client.register_script(self.RELEASE_LOCK_LUA)
        return bool(script(keys=[f"lock:{resource}"], args=[token]))
    
    # ==================== 任务队列实现 ====================
    
    def enqueue_task(self, queue: str, task_id: str, priority: int = 0) -> int:
        """入队任务 - 使用Sorted Set实现优先级队列"""
        return self.client.zadd(f"queue:{queue}", {task_id: priority})
    
    def dequeue_task(self, queue: str, timeout: int = 0) -> Optional[str]:
        """出队任务 - 阻塞式"""
        if timeout > 0:
            result = self.client.bzpopmin(f"queue:{queue}", timeout=timeout)
            if result:
                _, task_id, _ = result
                return task_id
        else:
            result = self.client.zpopmin(f"queue:{queue}", count=1)
            if result:
                return result[0][0]
        return None
    
    def get_queue_stats(self, queue: str) -> dict:
        """获取队列统计信息"""
        pipe = self.client.pipeline()
        pipe.zcard(f"queue:{queue}")
        pipe.zcount(f"queue:{queue}", '-inf', '+inf')
        pipe.zrange(f"queue:{queue}", 0, -1, withscores=True)
        counts = pipe.execute()
        
        return {
            "total": counts[0],
            "pending": counts[1],
            "tasks": [{"task_id": t[0], "priority": t[1]} for t in counts[2]]
        }

import etcd3
from typing import Any, Optional, List
from dataclasses import dataclass
import json
import time

@dataclass
class ServiceInstance:
    """服务实例"""
    service_name: str
    instance_id: str
    host: str
    port: int
    metadata: dict
    lease_id: str = None

class EtcdStateManager:
    """基于etcd的状态管理器 - 用于配置管理和服务发现"""
    
    def __init__(self, hosts: List[str], port: int = 2379):
        self.hosts = hosts
        self.port = port
        self.client = etcd3.client(hosts=hosts, port=port)
        self._lease_id = None
    
    # ==================== 租约管理 ====================
    
    def create_lease(self, ttl: int) -> str:
        """创建租约"""
        lease_id = self.client.lease(ttl=ttl)
        self._lease_id = lease_id
        return lease_id
    
    def keep_alive_lease(self, lease_id: str) -> bool:
        """保持租约活跃"""
        try:
            self.client.refresh_lease(lease_id)
            return True
        except Exception:
            return False
    
    # ==================== 键值操作 ====================
    
    def put(self, key: str, value: Any, lease_id: str = None) -> bool:
        """写入键值对"""
        try:
            if isinstance(value, (dict, list)):
                value = json.dumps(value)
            self.client.put(key, value, lease_id=lease_id)
            return True
        except Exception as e:
            print(f"Put failed: {e}")
            return False
    
    def get(self, key: str) -> Optional[Any]:
        """读取键值对"""
        try:
            value, _ = self.client.get(key)
            if value:
                try:
                    return json.loads(value)
                except json.JSONDecodeError:
                    return value.decode() if isinstance(value, bytes) else value
            return None
        except Exception:
            return None
    
    def delete(self, key: str) -> bool:
        """删除键值对"""
        try:
            self.client.delete(key)
            return True
        except Exception:
            return False
    
    def get_prefix(self, prefix: str) -> List[tuple[str, Any]]:
        """获取指定前缀的所有键值对"""
        try:
            results = []
            for value, metadata in self.client.get_prefix(prefix):
                if value:
                    key = metadata.key.decode() if isinstance(metadata.key, bytes) else metadata.key
                    try:
                        val = json.loads(value)
                    except json.JSONDecodeError:
                        val = value.decode() if isinstance(value, bytes) else value
                    results.append((key, val))
            return results
        except Exception:
            return []
    
    # ==================== 监视机制 ====================
    
    def watch_key(self, key: str, callback):
        """监视单个键的变化"""
        events_iterator, cancel = self.client.watch(key)
        try:
            for event in events_iterator:
                callback(event)
        finally:
            cancel()
    
    def watch_prefix(self, prefix: str, callback):
        """监视前缀下所有键的变化"""
        events_iterator, cancel = self.client.watch_prefix(prefix)
        try:
            for event in events_iterator:
                callback(event)
        finally:
            cancel()
    
    # ==================== 分布式锁 ====================
    
    def lock(self, key: str, value: str, ttl: int = 30) -> Optional[str]:
        """获取分布式锁（基于etcd的选举机制）"""
        lock = self.client.lock(key, ttl)
        if lock.acquire():
            return lock.key
        return None
    
    def unlock(self, lock) -> bool:
        """释放分布式锁"""
        return lock.release()
    
    # ==================== Leader选举 ====================
    
    def elect_leader(self, leader_name: str, ttl: int = 10) -> bool:
        """参与Leader选举"""
        try:
            # 使用自定义Leader选举
            election_key = f"/election/{leader_name}"
            current = self.get(election_key)
            
            if current is None or current['leader'] != leader_name:
                self.put(election_key, {
                    'leader': leader_name,
                    'timestamp': time.time()
                }, lease_id=self._lease_id)
            return True
        except Exception:
            return False
    
    def get_leader(self, election_name: str) -> Optional[str]:
        """获取当前Leader"""
        election_key = f"/election/{election_name}"
        result = self.get(election_key)
        if result:
            return result.get('leader')
        return None

import asyncio
from abc import ABC, abstractmethod
from typing import Any, List, Optional
from dataclasses import dataclass
from enum import Enum
import time
import hashlib

class ReplicationMode(Enum):
    SYNCHRONOUS = "sync"
    ASYNCHRONOUS = "async"
    SEMI_SYNCHRONOUS = "semi_sync"

@dataclass
class WriteResult:
    """写入结果"""
    success: bool
    term: int
    index: int
    message: str

class ReplicatedStateMachine:
    """复制状态机 - 主从复制实现"""
    
    def __init__(
        self, 
        node_id: str,
        peers: List[str],
        mode: ReplicationMode = ReplicationMode.SEMI_SYNCHRONOUS
    ):
        self.node_id = node_id
        self.peers = peers
        self.mode = mode
        
        # 状态机状态
        self.current_term = 0
        self.voted_for = None
        self.commit_index = 0
        self.last_applied = 0
        
        # 日志
        self.log: List[dict] = []
        
        # 伙伴节点连接
        self.peer_connections: dict[str, 'PeerConnection'] = {}
        
        # 状态存储
        self.state: dict[str, Any] = {}
        
        # 复制因子
        self.replication_factor = len(peers) + 1
    
    async def append_entries(
        self, 
        entries: List[dict],
        leader_id: str,
        prev_log_index: int,
        prev_log_term: int
    ) -> bool:
        """追加日志条目（从节点处理Leader的请求）"""
        # 1. 心跳检测
        if not entries:
            return True
        
        # 2. 一致性检查
        if prev_log_index > 0:
            if prev_log_index > len(self.log):
                return False
            if self.log[prev_log_index - 1]['term'] != prev_log_term:
                return False
        
        # 3. 追加新条目
        for i, entry in enumerate(entries):
            log_index = prev_log_index + i + 1
            if log_index <= len(self.log):
                # 冲突：替换现有条目
                self.log[log_index - 1] = entry
            else:
                self.log.append(entry)
        
        # 4. 更新提交索引
        self.commit_index = len(self.log)
        
        return True
    
    async def replicate_to_followers(self, entry: dict) -> WriteResult:
        """向所有从节点复制日志"""
        # 1. 本节点追加日志
        entry['term'] = self.current_term
        entry['index'] = len(self.log) + 1
        self.log.append(entry)
        
        # 2. 并发复制到从节点
        if self.mode == ReplicationMode.SYNCHRONOUS:
            success_count = await self._sync_replicate(entry)
            if success_count < self.replication_factor:
                return WriteResult(False, self.current_term, entry['index'], "Insufficient replicas")
        
        elif self.mode == ReplicationMode.ASYNC:
            # 异步复制，不等待结果
            asyncio.create_task(self._async_replicate(entry))
        
        elif self.mode == ReplicationMode.SEMI_SYNCHRONOUS:
            success_count = await self._sync_replicate(entry, min_acks=1)
            if success_count < 1:
                return WriteResult(False, self.current_term, entry['index'], "No ack from followers")
        
        # 3. 应用到状态机
        await self._apply_to_state_machine(entry)
        
        return WriteResult(True, self.current_term, entry['index'], "Success")
    
    async def _sync_replicate(self, entry: dict, min_acks: int = None) -> int:
        """同步复制到所有从节点"""
        if min_acks is None:
            min_acks = len(self.peers)
        
        acks = 0
        tasks = []
        
        for peer in self.peers:
            task = asyncio.create_task(self._send_append_entries(peer, entry))
            tasks.append(task)
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        for result in results:
            if result is True:
                acks += 1
        
        return acks
    
    async def _async_replicate(self, entry: dict) -> None:
        """异步复制，不等待结果"""
        for peer in self.peers:
            try:
                await self._send_append_entries(peer, entry)
            except Exception as e:
                print(f"Async replicate failed to {peer}: {e}")
    
    async def _send_append_entries(self, peer: str, entry: dict) -> bool:
        """发送AppendEntries RPC到从节点"""
        prev_log_index = entry['index'] - 1
        prev_log_term = self.log[prev_log_index - 1]['term'] if prev_log_index > 0 else 0
        
        try:
            # 模拟RPC调用
            # 在实际实现中，这里会调用peer的API
            conn = self.peer_connections.get(peer)
            if conn:
                return await conn.append_entries(
                    term=self.current_term,
                    leader_id=self.node_id,
                    prev_log_index=prev_log_index,
                    prev_log_term=prev_log_term,
                    entries=[entry],
                    leader_commit=self.commit_index
                )
        except Exception:
            return False
        return False
    
    async def _apply_to_state_machine(self, entry: dict) -> None:
        """将已提交的日志应用到状态机"""
        if entry['type'] == 'SET':
            self.state[entry['key']] = entry['value']
        elif entry['type'] == 'DELETE':
            self.state.pop(entry['key'], None)
        
        self.last_applied = entry['index']
    
    def get(self, key: str) -> Optional[Any]:
        """读取状态（可配置从哪个节点读取）"""
        return self.state.get(key)
    
    async def set(self, key: str, value: Any) -> WriteResult:
        """写入状态"""
        entry = {
            'type': 'SET',
            'key': key,
            'value': value,
            'timestamp': time.time()
        }
        return await self.replicate_to_followers(entry)

from typing import Any, Dict, Set
import time
import uuid

@dataclass
class VersionVector:
    """版本向量：跟踪每个节点的逻辑时钟"""
    versions: Dict[str, int]  # node_id -> version
    
    def increment(self, node_id: str) -> 'VersionVector':
        """递增当前节点的版本"""
        new_versions = self.versions.copy()
        new_versions[node_id] = new_versions.get(node_id, 0) + 1
        return VersionVector(new_versions)
    
    def merge(self, other: 'VersionVector') -> 'VersionVector':
        """合并两个版本向量：取每个分量的最大值"""
        all_nodes = set(self.versions.keys()) | set(other.versions.keys())
        merged = {
            node: max(self.versions.get(node, 0), other.versions.get(node, 0))
            for node in all_nodes
        }
        return VersionVector(merged)
    
    def happens_before(self, other: 'VersionVector') -> bool:
        """判断self是否在other之前（所有分量都<=）"""
        for node in set(self.versions.keys()) | set(other.versions.keys()):
            if self.versions.get(node, 0) > other.versions.get(node, 0):
                return False
        return True
    
    def is_concurrent(self, other: 'VersionVector') -> bool:
        """判断两个版本向量是否并发（既不happens_before对方）"""
        return not self.happens_before(other) and not other.happens_before(self)


@dataclass
class LWWRegister:
    """Last-Write-Wins Register - 最终写入胜出"""
    value: Any
    timestamp: float
    node_id: str
    
    @classmethod
    def create(cls, value: Any, node_id: str) -> 'LWWRegister':
        return cls(value, time.time(), node_id)
    
    def merge(self, other: 'LWWRegister') -> 'LWWRegister':
        """合并：时间戳大的胜出"""
        if other.timestamp > self.timestamp:
            return other
        return self


class GCounter:
    """G-Counter：只增计数器（CRDT）"""
    
    def __init__(self, node_id: str):
        self.node_id = node_id
        self.counters: Dict[str, int] = {node_id: 0}
    
    def increment(self, amount: int = 1) -> None:
        """本节点递增"""
        self.counters[self.node_id] = self.counters.get(self.node_id, 0) + amount
    
    def value(self) -> int:
        """返回总计数（所有节点之和）"""
        return sum(self.counters.values())
    
    def merge(self, other: 'GCounter') -> None:
        """合并：取每个分量的最大值"""
        for node, count in other.counters.items():
            self.counters[node] = max(self.counters.get(node, 0), count)


class MultiMasterState:
    """多主状态管理器"""
    
    def __init__(self, node_id: str, cluster_nodes: List[str]):
        self.node_id = node_id
        self.cluster_nodes = cluster_nodes
        
        # 状态存储
        self.state: Dict[str, LWWRegister] = {}
        
        # 版本向量
        self.version_vector = VersionVector({node_id: 0})
        
        # 待同步的写操作队列
        self.pending_writes: List[tuple[VersionVector, str, Any]] = []
    
    async def local_write(self, key: str, value: Any) -> None:
        """本地写入"""
        # 1. 更新本地状态
        register = LWWRegister.create(value, self.node_id)
        self.state[key] = register
        
        # 2. 递增版本向量
        self.version_vector = self.version_vector.increment(self.node_id)
        
        # 3. 加入待同步队列
        self.pending_writes.append((self.version_vector, key, register))
    
    async def merge_remote_state(
        self, 
        remote_version: VersionVector, 
        key: str, 
        remote_register: LWWRegister
    ) -> bool:
        """合并远程状态"""
        # 1. 检测冲突
        local_register = self.state.get(key)
        
        if local_register is None:
            # 无本地状态，直接采纳
            self.state[key] = remote_register
            self.version_vector = self.version_vector.merge(remote_version)
            return True
        
        # 2. 判断关系
        if remote_version.happens_before(self.version_vector):
            # 远程状态落后，无需合并
            return False
        elif self.version_vector.happens_before(remote_version):
            # 本地状态落后，采纳远程
            self.state[key] = remote_register
            self.version_vector = self.version_vector.merge(remote_version)
            return True
        else:
            # 并发冲突：使用LWW解决
            winner = local_register.merge(remote_register)
            self.state[key] = winner
            self.version_vector = self.version_vector.merge(remote_version)
            return True
    
    async def sync_with_peer(self, peer_id: str) -> None:
        """与对等节点同步状态"""
        # 1. 发送本地版本向量和待同步的写操作
        # 2. 接收对方的版本向量和写操作
        # 3. 执行合并
        
        # 这是简化的伪代码
        # 实际实现需要网络通信
        pass

强一致 → 顺序一致 → 因果一致 → 入口一致 → 最终一致 → 弱一致
   ↑                                                          ↓
   └────────────────── 可调一致性 ────────────────────────────┘

from enum import Enum
from dataclasses import dataclass, field
from typing import Any, Optional, List, Dict
import time
import asyncio
import random

class NodeState(Enum):
    FOLLOWER = "follower"
    CANDIDATE = "candidate"
    LEADER = "leader"

@dataclass
class LogEntry:
    """日志条目"""
    term: int
    index: int
    command: Any
    timestamp: float = field(default_factory=time.time)

@dataclass
class VoteRequest:
    """投票请求"""
    term: int
    candidate_id: str
    last_log_index: int
    last_log_term: int

@dataclass
class VoteResponse:
    """投票响应"""
    term: int
    vote_granted: bool

@dataclass
class AppendEntriesRequest:
    """追加日志请求"""
    term: int
    leader_id: str
    prev_log_index: int
    prev_log_term: int
    entries: List[LogEntry]
    leader_commit: int

@dataclass
class AppendEntriesResponse:
    """追加日志响应"""
    term: int
    success: bool
    match_index: int = 0

class RaftNode:
    """Raft共识算法实现节点"""
    
    def __init__(self, node_id: str, peers: List[str]):
        self.node_id = node_id
        self.peers = peers
        
        # 持久化状态（实际应持久化到磁盘）
        self.current_term = 0
        self.voted_for: Optional[str] = None
        self.log: List[LogEntry] = []
        
        # 易失状态
        self.state = NodeState.FOLLOWER
        self.commit_index = 0
        self.last_applied = 0
        
        # Leader专用易失状态
        self.next_index: Dict[str, int] = {}
        self.match_index: Dict[str, int] = {}
        
        # 选举超时
        self.election_timeout = self._random_election_timeout()
        self.last_heartbeat = time.time()
        
        # 状态机
        self.state_machine: Dict[str, Any] = {}
        
        # 网络层（简化）
        self.network: Dict[str, 'RaftNode'] = {}
    
    def _random_election_timeout(self) -> float:
        """生成随机选举超时（150-300ms）"""
        return 0.15 + random.random() * 0.15
    
    # ==================== Leader选举 ====================
    
    async def start_election(self) -> None:
        """开始选举"""
        self.state = NodeState.CANDIDATE
        self.current_term += 1
        self.voted_for = self.node_id  # 给自己投票
        
        votes = 1  # 自己的票
        
        # 并发请求所有节点投票
        vote_requests = [
            self._request_vote(peer)
            for peer in self.peers
        ]
        
        for response in asyncio.as_completed(vote_requests):
            if self.state != NodeState.CANDIDATE:
                break  # 已经不是Candidate
            
            result = await response
            if result.vote_granted:
                votes += 1
                if votes > (len(self.peers) + 1) // 2:
                    # 获得多数票，成为Leader
                    await self.become_leader()
                    return
    
    async def _request_vote(self, peer_id: str) -> VoteResponse:
        """发送投票请求"""
        request = VoteRequest(
            term=self.current_term,
            candidate_id=self.node_id,
            last_log_index=self._last_log_index(),
            last_log_term=self._last_log_term()
        )
        
        # 模拟RPC
        peer = self.network.get(peer_id)
        if peer:
            return await peer.request_vote(request)
        return VoteResponse(term=self.current_term, vote_granted=False)
    
    async def request_vote(self, request: VoteRequest) -> VoteResponse:
        """处理投票请求"""
        # 1. 任期检查
        if request.term > self.current_term:
            self.current_term = request.term
            self.state = NodeState.FOLLOWER
            self.voted_for = None
        
        # 2. 投票条件检查
        can_vote = (
            self.voted_for is None or 
            self.voted_for == request.candidate_id or
            request.term > self.current_term
        )
        
        log_ok = (
            request.last_log_term > self._last_log_term() or
            (request.last_log_term == self._last_log_term() and 
             request.last_log_index >= self._last_log_index())
        )
        
        if can_vote and log_ok:
            self.voted_for = request.candidate_id
            return VoteResponse(term=self.current_term, vote_granted=True)
        
        return VoteResponse(term=self.current_term, vote_granted=False)
    
    async def become_leader(self) -> None:
        """成为Leader"""
        self.state = NodeState.LEADER
        
        # 初始化Leader状态
        for peer in self.peers:
            self.next_index[peer] = self._last_log_index() + 1
            self.match_index[peer] = 0
        
        # 立即发送心跳
        asyncio.create_task(self._send_heartbeats())
    
    async def _send_heartbeats(self) -> None:
        """发送心跳"""
        while self.state == NodeState.LEADER:
            for peer in self.peers:
                await self._send_append_entries(peer)
            await asyncio.sleep(0.05)  # 心跳间隔50ms
    
    # ==================== 日志复制 ====================
    
    async def append_entries(
        self, 
        request: AppendEntriesRequest
    ) -> AppendEntriesResponse:
        """处理追加日志请求"""
        # 1. 任期检查
        if request.term < self.current_term:
            return AppendEntriesResponse(term=self.current_term, success=False)
        
        if request.term > self.current_term:
            self.current_term = request.term
            self.state = NodeState.FOLLOWER
        
        self.last_heartbeat = time.time()
        
        # 2. 一致性检查
        if request.prev_log_index > 0:
            if request.prev_log_index > len(self.log):
                return AppendEntriesResponse(term=self.current_term, success=False)
            if self.log[request.prev_log_index - 1].term != request.prev_log_term:
                return AppendEntriesResponse(term=self.current_term, success=False)
        
        # 3. 追加新条目
        for i, entry in enumerate(request.entries):
            log_index = request.prev_log_index + i + 1
            if log_index <= len(self.log):
                # 覆盖冲突条目
                self.log[log_index - 1] = entry
            else:
                self.log.append(entry)
        
        # 4. 更新提交索引
        if request.leader_commit > self.commit_index:
            self.commit_index = min(request.leader_commit, len(self.log))
        
        return AppendEntriesResponse(term=self.current_term, success=True)
    
    async def _send_append_entries(self, peer_id: str) -> None:
        """发送追加日志请求到从节点"""
        peer = self.network.get(peer_id)
        if not peer:
            return
        
        next_idx = self.next_index[peer_id]
        prev_log_index = next_idx - 1
        prev_log_term = self.log[prev_log_index - 1].term if prev_log_index > 0 else 0
        
        # 获取要发送的条目
        entries = self.log[prev_log_index:] if prev_log_index < len(self.log) else []
        
        request = AppendEntriesRequest(
            term=self.current_term,
            leader_id=self.node_id,
            prev_log_index=prev_log_index,
            prev_log_term=prev_log_term,
            entries=entries,
            leader_commit=self.commit_index
        )
        
        response = await peer.append_entries(request)
        
        if response.success:
            # 更新next_index和match_index
            self.next_index[peer_id] = len(self.log) + 1
            self.match_index[peer_id] = len(self.log)
        else:
            # 递减next_index重试
            self.next_index[peer_id] -= 1
    
    # ==================== 客户端接口 ====================
    
    async def propose(self, command: Any) -> bool:
        """客户端提交命令"""
        if self.state != NodeState.LEADER:
            return False
        
        # 1. 创建日志条目
        entry = LogEntry(
            term=self.current_term,
            index=len(self.log) + 1,
            command=command
        )
        self.log.append(entry)
        
        # 2. 复制到多数节点
        await self._replicate_to_majority(entry)
        
        # 3. 提交日志
        self.commit_index = entry.index
        await self._apply_to_state_machine()
        
        return True
    
    async def _replicate_to_majority(self, entry: LogEntry) -> None:
        """复制到多数节点"""
        acks = 1
        for peer in self.peers:
            if self.match_index.get(peer, 0) >= entry.index:
                acks += 1
        
        if acks > (len(self.peers) + 1) // 2:
            return
        
        # 发送AppendEntries
        tasks = [self._send_append_entries(peer) for peer in self.peers]
        await asyncio.gather(*tasks, return_exceptions=True)
    
    async def _apply_to_state_machine(self) -> None:
        """应用已提交日志到状态机"""
        while self.last_applied < self.commit_index:
            self.last_applied += 1
            entry = self.log[self.last_applied - 1]
            
            # 执行命令
            if isinstance(entry.command, dict):
                cmd_type = entry.command.get('type')
                if cmd_type == 'SET':
                    self.state_machine[entry.command['key']] = entry.command['value']
                elif cmd_type == 'DELETE':
                    self.state_machine.pop(entry.command['key'], None)
    
    def _last_log_index(self) -> int:
        return len(self.log)
    
    def _last_log_term(self) -> int:
        if self.log:
            return self.log[-1].term
        return 0

import hashlib
import asyncio
from typing import Any, Dict, Set, List
from dataclasses import dataclass
import random

@dataclass
class MerkleNode:
    """Merkle树节点"""
    hash: str
    left: 'MerkleNode' = None
    right: 'MerkleNode' = None
    value: Any = None  # 叶节点存储的值

class MerkleTree:
    """Merkle树 - 高效的数据一致性校验"""
    
    def __init__(self, data: Dict[str, Any]):
        self.root = self._build_tree(data)
    
    def _hash_value(self, key: str, value: Any) -> str:
        """计算键值对的哈希"""
        data = f"{key}:{str(value)}"
        return hashlib.sha256(data.encode()).hexdigest()[:16]
    
    def _build_tree(self, data: Dict[str, Any]) -> MerkleNode:
        """构建Merkle树"""
        if not data:
            return MerkleNode(hash="0" * 16)
        
        # 创建叶节点
        leaves = [
            MerkleNode(hash=self._hash_value(k, v), value=v)
            for k, v in data.items()
        ]
        
        # 如果只有一个叶节点，直接返回
        if len(leaves) == 1:
            return leaves[0]
        
        # 补齐到偶数个叶节点
        if len(leaves) % 2 != 0:
            leaves.append(MerkleNode(hash="0" * 16))
        
        # 两两合并构建树
        while len(leaves) > 1:
            parents = []
            for i in range(0, len(leaves), 2):
                left = leaves[i]
                right = leaves[i + 1]
                combined_hash = hashlib.sha256(
                    (left.hash + right.hash).encode()
                ).hexdigest()[:16]
                parents.append(MerkleNode(hash=combined_hash, left=left, right=right))
            leaves = parents
        
        return leaves[0]
    
    def diff(self, other: 'MerkleTree') -> Set[str]:
        """计算两棵Merkle树的差异（键集合）"""
        diff_keys = set()
        self._diff_recursive(self.root, other.root, diff_keys)
        return diff_keys
    
    def _diff_recursive(
        self, 
        node1: MerkleNode, 
        node2: MerkleNode, 
        diff: Set[str]
    ) -> None:
        """递归比较差异"""
        if node1.hash == node2.hash:
            return  # 完全相同
        
        if node1.value is not None and node2.value is not None:
            # 都是叶节点且不同
            # 这里需要外部映射来知道是哪个键
            diff.add("modified")
            return
        
        if node1.left and node2.left:
            self._diff_recursive(node1.left, node2.left, diff)
        if node1.right and node2.right:
            self._diff_recursive(node1.right, node2.right, diff)


class GossipProtocol:
    """Gossip协议实现 - 最终一致性传播"""
    
    def __init__(self, node_id: str, peers: List[str]):
        self.node_id = node_id
        self.peers = list(peers)
        
        # 本地状态
        self.state: Dict[str, Any] = {}
        self.version: Dict[str, int] = {}
        
        # 摘要信息（用于反熵）
        self.summary: Dict[str, str] = {}  # key -> hash
        
        # 配置
        self.gossip_interval = 1.0  # 秒
        self.gossip_fanout = 3  # 每次选择多少个节点
    
    async def update(self, key: str, value: Any) -> None:
        """更新本地状态"""
        self.state[key] = value
        self.version[key] = self.version.get(key, 0) + 1
        self.summary[key] = hashlib.sha256(
            f"{key}:{value}:{self.version[key]}".encode()
        ).hexdigest()[:16]
    
    async def gossip_loop(self) -> None:
        """Gossip主循环"""
        while True:
            await asyncio.sleep(self.gossip_interval)
            
            # 1. 选择要同步的节点
            targets = random.sample(
                self.peers, 
                min(self.gossip_fanout, len(self.peers))
            )
            
            # 2. 发送摘要
            for target in targets:
                await self._send_summary(target)
    
    async def _send_summary(self, target: str) -> None:
        """发送本地摘要到目标节点"""
        # 实际实现中通过网络发送
        # 这里模拟：发送摘要，接收差异
        pass
    
    async def merge_state(self, remote_state: Dict[str, Any]) -> int:
        """合并远程状态（使用版本向量）"""
        merged_count = 0
        
        for key, remote_value in remote_state.items():
            local_version = self.version.get(key, 0)
            # 假设remote_value包含版本信息
            remote_version = remote_value.get('_version', 0) if isinstance(remote_value, dict) else 0
            
            if remote_version > local_version:
                self.state[key] = remote_value.get('value', remote_value)
                self.version[key] = remote_version
                merged_count += 1
        
        return merged_count

import time
import uuid
import asyncio
from typing import List, Optional
import redis

class Redlock:
    """Redlock分布式锁实现"""
    
    def __init__(self, redis_hosts: List[tuple], retry_count: int = 3, retry_delay: float = 0.2):
        """
        初始化Redlock
        
        Args:
            redis_hosts: Redis实例列表 [(host, port), ...]
            retry_count: 重试次数
            retry_delay: 重试延迟（秒）
        """
        self.redis_hosts = redis_hosts
        self.retry_count = retry_count
        self.retry_delay = retry_delay
        self.n = len(redis_hosts)
        self.quorum = self.n // 2 + 1
        
        # 创建Redis连接
        self.clients = [
            redis.Redis(host=host, port=port, socket_timeout=5)
            for host, port in redis_hosts
        ]
        
        # 锁配置
        self.clock_drift_factor = 0.01  # 时钟漂移因子
        self.auto_release_time = 10000  # 默认10秒
    
    def _generate_token(self) -> str:
        """生成唯一的锁token"""
        return str(uuid.uuid4())
    
    async def acquire(
        self, 
        resource: str, 
        ttl_ms: int = None
    ) -> Optional[str]:
        """
        获取分布式锁
        
        Args:
            resource: 资源名称
            ttl_ms: 锁过期时间（毫秒）
            
        Returns:
            成功返回token，失败返回None
        """
        if ttl_ms is None:
            ttl_ms = self.auto_release_time
        
        token = self._generate_token()
        
        for attempt in range(self.retry_count):
            acquired = 0
            start_time = int(time.time() * 1000)
            
            # 并发向所有实例获取锁
            for client in self.clients:
                if await self._lock_instance(client, resource, token, ttl_ms):
                    acquired += 1
            
            # 计算获取锁的总耗时
            elapsed = int(time.time() * 1000) - start_time
            
            # 计算锁的实际有效时间
            validity_time = ttl_ms - elapsed - (ttl_ms * self.clock_drift_factor)
            
            if acquired >= self.quorum and validity_time > 0:
                return token
            
            # 获取失败，释放已获取的锁
            await self._release_all(resource, token)
            
            # 重试前等待
            if attempt < self.retry_count - 1:
                await asyncio.sleep(self.retry_delay)
        
        return None
    
    async def _lock_instance(
        self, 
        client: redis.Redis, 
        resource: str, 
        token: str, 
        ttl_ms: int
    ) -> bool:
        """向单个Redis实例获取锁"""
        try:
            # SET key value NX PX timeout
            result = await asyncio.to_thread(
                client.set,
                f"lock:{resource}",
                token,
                nx=True,
                px=ttl_ms
            )
            return result is True
        except Exception:
            return False
    
    async def release(self, resource: str, token: str) -> bool:
        """释放分布式锁"""
        await self._release_all(resource, token)
        return True
    
    async def _release_all(self, resource: str, token: str) -> None:
        """向所有实例释放锁"""
        tasks = [
            self._unlock_instance(client, resource, token)
            for client in self.clients
        ]
        await asyncio.gather(*tasks, return_exceptions=True)
    
    async def def _unlock_instance(
        self, 
        client: redis.Redis, 
        resource: str, 
        token: str
    ) -> bool:
        """向单个Redis实例释放锁"""
        # 使用Lua脚本保证原子性
        lua_script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        try:
            result = await asyncio.to_thread(
                client.eval,
                lua_script,
                1,
                f"lock:{resource}",
                token
            )
            return result == 1
        except Exception:
            return False
    
    async def extend(
        self, 
        resource: str, 
        token: str, 
        additional_ttl_ms: int
    ) -> bool:
        """延长锁的持有时间"""
        lua_script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("pexpire", KEYS[1], ARGV[2])
        else
            return 0
        end
        """
        results = []
        for client in self.clients:
            try:
                result = await asyncio.to_thread(
                    client.eval,
                    lua_script,
                    1,
                    f"lock:{resource}",
                    token,
                    additional_ttl_ms
                )
                results.append(result == 1)
            except Exception:
                results.append(False)
        
        return sum(results) >= self.quorum


# ==================== 简化版单Redis锁 ====================

class SimpleDistributedLock:
    """简化版分布式锁（单Redis实例）"""
    
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.local_locks: dict = {}  # 本地锁追踪
    
    async def acquire(
        self, 
        resource: str, 
        timeout: float = 10.0,
        ttl_ms: int = 30000
    ) -> Optional[str]:
        """获取锁（带超时）"""
        token = str(uuid.uuid4())
        end_time = time.time() + timeout
        
        while time.time() < end_time:
            # 尝试获取锁
            acquired = await asyncio.to_thread(
                self.redis.set,
                f"lock:{resource}",
                token,
                nx=True,
                px=ttl_ms
            )
            
            if acquired:
                return token
            
            await asyncio.sleep(0.001)  # 避免CPU空转
        
        return None
    
    async def release(self, resource: str, token: str) -> bool:
        """释放锁"""
        lua_script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        result = await asyncio.to_thread(
            self.redis.eval,
            lua_script,
            1,
            f"lock:{resource}",
            token
        )
        return result == 1
    
    async def __aenter__(self) -> 'SimpleDistributedLock':
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        # 上下文管理器支持
        pass

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

@dataclass
class FencingToken:
    """Fencing令牌"""
    token: int
    resource: str
    owner: str
    issued_at: float
    
    def is_valid(self, current_token: int) -> bool:
        """验证token是否有效"""
        return self.token > current_token


class FencingTokenService:
    """Fencing Token服务 - 解决分布式锁的安全性问题"""
    
    def __init__(self):
        self.tokens: dict[str, int] = {}  # resource -> current token
        self.token_owners: dict[str, FencingToken] = {}  # token -> metadata
    
    def issue_token(self, resource: str, owner: str) -> FencingToken:
        """发放新的fencing token"""
        current = self.tokens.get(resource, 0)
        new_token = current + 1
        
        token = FencingToken(
            token=new_token,
            resource=resource,
            owner=owner,
            issued_at=time.time()
        )
        
        self.tokens[resource] = new_token
        self.token_owners[f"{resource}:{new_token}"] = token
        
        return token
    
    def validate_token(self, token: FencingToken) -> bool:
        """验证token是否有效"""
        current = self.tokens.get(token.resource, 0)
        return token.token > current
    
    def get_current_token(self, resource: str) -> int:
        """获取资源的当前token"""
        return self.tokens.get(resource, 0)


class FencingStorageClient:
    """支持Fencing Token的存储客户端"""
    
    def __init__(self, storage_backend, token_service: FencingTokenService):
        self.storage = storage_backend
        self.token_service = token_service
    
    async def write_with_token(
        self, 
        resource: str, 
        data: Any, 
        token: FencingToken
    ) -> bool:
        """使用Fencing Token写入数据"""
        # 1. 验证token
        if not self.token_service.validate_token(token):
            raise ValueError(f"Invalid token: {token.token}")
        
        # 2. 写入数据（带token用于乐观锁）
        return await self.storage.update(
            resource, 
            data, 
            expected_token=token.token
        )


class FencingAwareLock:
    """集成Fencing Token的分布式锁"""
    
    def __init__(self, lock: SimpleDistributedLock, token_service: FencingTokenService):
        self.lock = lock
        self.token_service = token_service
        self.current_token: Optional[FencingToken] = None
        self.resource: Optional[str] = None
    
    async def acquire(self, resource: str, **kwargs) -> Optional[FencingToken]:
        """获取锁并返回fencing token"""
        token = await self.lock.acquire(resource, **kwargs)
        if token:
            self.resource = resource
            self.current_token = self.token_service.issue_token(resource, token)
            return self.current_token
        return None
    
    async def release(self) -> bool:
        """释放锁"""
        if self.current_token and self.resource:
            return await self.lock.release(self.resource, self.current_token.owner)
        return False
    
    async def __aenter__(self) -> 'FencingAwareLock':
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        await self.release()

class LockFreeCounter:
    """无锁计数器 - 乐观并发控制示例"""
    
    def __init__(self, storage, key: str):
        self.storage = storage
        self.key = key
    
    async def increment(self, expected_version: int) -> tuple[bool, int]:
        """
        乐观锁递增
        
        Returns:
            (success, new_value)
        """
        # 读取当前值
        current = await self.storage.get(self.key)
        current_value = current['value']
        current_version = current['version']
        
        # 版本检查
        if current_version != expected_version:
            return False, current_value
        
        # CAS更新
        new_value = current_value + 1
        success = await self.storage.compare_and_set(
            self.key,
            expected_value=current_value,
            new_value=new_value,
            expected_version=current_version
        )
        
        if success:
            return True, new_value
        return False, current_value


class IdempotentOperation:
    """幂等操作封装"""
    
    def __init__(self, storage, operation_id: str):
        self.storage = storage
        self.operation_id = operation_id
    
    async def execute(self, operation: callable, *args, **kwargs):
        """
        执行幂等操作
        
        操作结果会被缓存，相同operation_id的重复调用返回缓存结果
        """
        # 1. 检查是否已执行
        cached = await self.storage.get(f"result:{self.operation_id}")
        if cached:
            return cached['result']
        
        # 2. 执行操作
        result = await operation(*args, **kwargs)
        
        # 3. 缓存结果
        await self.storage.set(f"result:{self.operation_id}", {
            'result': result,
            'completed_at': time.time()
        })
        
        return result

# state_manager/models.py
from __future__ import annotations
import time
from typing import Any, Optional, Dict, List, Callable
from dataclasses import dataclass, field
from enum import Enum
import asyncio
import uuid

class OperationType(Enum):
    """状态操作类型"""
    GET = "get"
    SET = "set"
    DELETE = "delete"
    UPDATE = "update"
    WATCH = "watch"

@dataclass
class StateEntry:
    """状态条目"""
    key: str
    value: Any
    version: int
    created_at: float
    updated_at: float
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            'key': self.key,
            'value': self.value,
            'version': self.version,
            'created_at': self.created_at,
            'updated_at': self.updated_at,
            'metadata': self.metadata
        }

@dataclass 
class StateChange:
    """状态变更事件"""
    key: str
    old_value: Any
    new_value: Any
    version: int
    timestamp: float
    operation: OperationType
    metadata: Dict[str, Any] = field(default_factory=dict)

@dataclass
class ConflictError(Exception):
    """并发冲突错误"""
    key: str
    expected_version: int
    actual_version: int
    current_value: Any
    
    def __str__(self):
        return f"Conflict on key '{self.key}': expected version {self.expected_version}, actual {self.actual_version}"

@dataclass
class OptimisticLock:
    """乐观锁配置"""
    enabled: bool = True
    max_retries: int = 3
    retry_delay: float = 0.1

# state_manager/storage.py
from abc import ABC, abstractmethod
from typing import Any, Optional, List, Tuple
import asyncio
import aioredis
import json

class StorageBackend(ABC):
    """存储后端抽象基类"""
    
    @abstractmethod
    async def get(self, key: str) -> Optional[StateEntry]:
        pass
    
    @abstractmethod
    async def set(self, key: str, entry: StateEntry) -> StateEntry:
        pass
    
    @abstractmethod
    async def delete(self, key: str) -> bool:
        pass
    
    @abstractmethod
    async def compare_and_set(
        self, 
        key: str, 
        expected_version: int, 
        new_entry: StateEntry
    ) -> bool:
        pass
    
    @abstractmethod
    async def scan(self, pattern: str) -> List[StateEntry]:
        pass


class RedisStorage(StorageBackend):
    """Redis存储后端实现"""
    
    def __init__(self, redis_url: str):
        self.redis_url = redis_url
        self._client: Optional[aioredis.Redis] = None
    
    async def connect(self):
        self._client = await aioredis.create_redis_pool(self.redis_url)
    
    async def close(self):
        if self._client:
            self._client.close()
    
    def _entry_to_hash(self, entry: StateEntry) -> Dict[str, str]:
        return {
            'key': entry.key,
            'value': json.dumps(entry.value),
            'version': str(entry.version),
            'created_at': str(entry.created_at),
            'updated_at': str(entry.updated_at),
            'metadata': json.dumps(entry.metadata)
        }
    
    def _hash_to_entry(self, data: Dict[str, bytes]) -> StateEntry:
        return StateEntry(
            key=data[b'key'].decode(),
            value=json.loads(data[b'value']),
            version=int(data[b'version']),
            created_at=float(data[b'created_at']),
            updated_at=float(data[b'updated_at']),
            metadata=json.loads(data[b'metadata']) if b'metadata' in data else {}
        )
    
    async def get(self, key: str) -> Optional[StateEntry]:
        data = await self._client.hgetall(f'state:{key}')
        if not data:
            return None
        return self._hash_to_entry(data)
    
    async def set(self, key: str, entry: StateEntry) -> StateEntry:
        now = time.time()
        if entry.created_at == 0:
            entry.created_at = now
        entry.updated_at = now
        
        pipe = self._client.pipeline()
        pipe.hmset(f'state:{key}', self._entry_to_hash(entry))
        await pipe.execute()
        return entry
    
    async def delete(self, key: str) -> bool:
        result = await self._client.delete(f'state:{key}')
        return result > 0
    
    async def compare_and_set(
        self, 
        key: str, 
        expected_version: int, 
        new_entry: StateEntry
    ) -> bool:
        """原子性CAS操作"""
        lua_script = """
        local current = redis.call('HGET', KEYS[1], 'version')
        if current == false then
            return -1  -- key不存在
        end
        if tonumber(current) ~= tonumber(ARGV[1]) then
            return 0  -- 版本不匹配
        end
        -- 版本匹配，执行更新
        redis.call('HMSET', KEYS[1], unpack(ARGV[2]))
        return 1
        """
        
        argv = [
            str(expected_version),
            'key', new_entry.key,
            'value', json.dumps(new_entry.value),
            'version', str(new_entry.version),
            'created_at', str(new_entry.created_at),
            'updated_at', str(time.time()),
            'metadata', json.dumps(new_entry.metadata)
        ]
        
        result = await self._client.eval(lua_script, 1, f'state:{key}', *argv)
        return result == 1
    
    async def scan(self, pattern: str) -> List[StateEntry]:
        results = []
        cursor = 0
        full_pattern = f'state:{pattern}'
        
        while True:
            cursor, keys = await self._client.scan(cursor, match=full_pattern, count=100)
            for key in keys:
                key_str = key.decode() if isinstance(key, bytes) else key
                entry = await self.get(key_str.replace('state:', ''))
                if entry:
                    results.append(entry)
            
            if cursor == 0:
                break
        
        return results


class MemoryStorage(StorageBackend):
    """内存存储后端（测试用）"""
    
    def __init__(self):
        self._store: Dict[str, StateEntry] = {}
        self._locks: Dict[str, asyncio.Lock] = {}
    
    def _get_lock(self, key: str) -> asyncio.Lock:
        if key not in self._locks:
            self._locks[key] = asyncio.Lock()
        return self._locks[key]
    
    async def get(self, key: str) -> Optional[StateEntry]:
        return self._store.get(key)
    
    async def set(self, key: str, entry: StateEntry) -> StateEntry:
        now = time.time()
        if entry.created_at == 0:
            entry.created_at = now
        entry.updated_at = now
        
        async with self._get_lock(key):
            self._store[key] = entry
        return entry
    
    async def delete(self, key: str) -> bool:
        async with self._get_lock(key):
            if key in self._store:
                del self._store[key]
                return True
        return False
    
    async def compare_and_set(
        self, 
        key: str, 
        expected_version: int, 
        new_entry: StateEntry
    ) -> bool:
        async with self._get_lock(key):
            current = self._store.get(key)
            
            if current is None:
                return False
            
            if current.version != expected_version:
                return False
            
            new_entry.created_at = current.created_at
            new_entry.updated_at = time.time()
            self._store[key] = new_entry
            return True
    
    async def scan(self, pattern: str) -> List[StateEntry]:
        import fnmatch
        results = []
        for key, entry in self._store.items():
            if fnmatch.fnmatch(key, pattern):
                results.append(entry)
        return results

# state_manager/manager.py
class StateManager:
    """分布式状态管理器 - 支持乐观锁"""
    
    def __init__(
        self,
        storage: StorageBackend,
        lock_config: OptimisticLock = None
    ):
        self.storage = storage
        self.lock_config = lock_config or OptimisticLock()
        
        # 观察者
        self._watchers: Dict[str, List[Callable[[StateChange], None]]] = {}
        
        # 指标
        self._metrics = {
            'gets': 0,
            'sets': 0,
            'conflicts': 0,
            'retries': 0
        }
    
    async def get(self, key: str) -> Optional[StateEntry]:
        """获取状态"""
        self._metrics['gets'] += 1
        return await self.storage.get(key)
    
    async def set(
        self,
        key: str,
        value: Any,
        version: int = None,
        metadata: Dict[str, Any] = None,
        use_ optimistic_lock: bool = None
    ) -> StateEntry:
        """
        设置状态
        
        Args:
            key: 状态键
            value: 状态值
            version: 期望的版本号（用于乐观锁）
            metadata: 元数据
            use_optimistic_lock: 是否使用乐观锁（None使用默认配置）
        """
        use_lock = use_optimistic_lock if use_optimistic_lock is not None else self.lock_config.enabled
        
        if use_lock and version is not None:
            return await self._set_with_optimistic_lock(key, value, version, metadata)
        else:
            return await self._set_simple(key, value, metadata)
    
    async def _set_simple(
        self, 
        key: str, 
        value: Any, 
        metadata: Dict[str, Any] = None
    ) -> StateEntry:
        """简单设置（无版本检查）"""
        self._metrics['sets'] += 1
        
        # 获取当前条目以继承版本号
        current = await self.storage.get(key)
        new_version = (current.version + 1) if current else 1
        
        entry = StateEntry(
            key=key,
            value=value,
            version=new_version,
            created_at=0,  # 由storage填充
            updated_at=0,
            metadata=metadata or {}
        )
        
        result = await self.storage.set(key, entry)
        
        # 触发观察者
        await self._notify_watchers(StateChange(
            key=key,
            old_value=current.value if current else None,
            new_value=value,
            version=result.version,
            timestamp=time.time(),
            operation=OperationType.SET,
            metadata=metadata or {}
        ))
        
        return result
    
    async def _set_with_optimistic_lock(
        self,
        key: str,
        value: Any,
        expected_version: int,
        metadata: Dict[str, Any] = None
    ) -> StateEntry:
        """使用乐观锁设置状态"""
        for attempt in range(self.lock_config.max_retries):
            # 获取当前值
            current = await self.storage.get(key)
            actual_version = current.version if current else 0
            
            # 版本检查
            if current is None and expected_version != 0:
                raise ConflictError(
                    key=key,
                    expected_version=expected_version,
                    actual_version=0,
                    current_value=None
                )
            
            if current and current.version != expected_version:
                self._metrics['conflicts'] += 1
                raise ConflictError(
                    key=key,
                    expected_version=expected_version,
                    actual_version=current.version,
                    current_value=current.value
                )
            
            # 构建新条目
            new_version = actual_version + 1
            entry = StateEntry(
                key=key,
                value=value,
                version=new_version,
                created_at=current.created_at if current else 0,
                updated_at=0,
                metadata=metadata or {}
            )
            
            # CAS更新
            success = await self.storage.compare_and_set(key, expected_version, entry)
            
            if success:
                self._metrics['sets'] += 1
                
                # 触发观察者
                await self._notify_watchers(StateChange(
                    key=key,
                    old_value=current.value if current else None,
                    new_value=value,
                    version=new_version,
                    timestamp=time.time(),
                    operation=OperationType.SET,
                    metadata=metadata or {}
                ))
                
                return entry
            else:
                self._metrics['retries'] += 1
                await asyncio.sleep(self.lock_config.retry_delay)
        
        raise ConflictError(
            key=key,
            expected_version=expected_version,
            actual_version=-1,
            current_value=None
        )
    
    async def update(
        self,
        key: str,
        updater: Callable[[Any], Any],
        metadata: Dict[str, Any] = None
    ) -> StateEntry:
        """更新状态（函数式）"""
        current = await self.get(key)
        expected_version = current.version if current else 0
        
        for attempt in range(self.lock_config.max_retries):
            if current is None:
                raise ValueError(f"Key '{key}' not found")
            
            try:
                new_value = updater(current.value)
            except Exception as e:
                raise ValueError(f"Updater function failed: {e}")
            
            new_version = current.version + 1
            entry = StateEntry(
                key=key,
                value=new_value,
                version=new_version,
                created_at=current.created_at,
                updated_at=0,
                metadata=metadata or current.metadata
            )
            
            success = await self.storage.compare_and_set(key, current.version, entry)
            
            if success:
                await self._notify_watchers(StateChange(
                    key=key,
                    old_value=current.value,
                    new_value=new_value,
                    version=new_version,
                    timestamp=time.time(),
                    operation=OperationType.UPDATE,
                    metadata=metadata or {}
                ))
                return entry
            
            # 更新失败，重试
            current = await self.storage.get(key)
            self._metrics['retries'] += 1
            await asyncio.sleep(self.lock_config.retry_delay)
        
        raise ConflictError(
            key=key,
            expected_version=expected_version,
            actual_version=current.version if current else -1,
            current_value=current.value if current else None
        )
    
    async def delete(self, key: str) -> bool:
        """删除状态"""
        current = await self.get(key)
        if current is None:
            return False
        
        success = await self.storage.delete(key)
        
        if success:
            await self._notify_watchers(StateChange(
                key=key,
                old_value=current.value,
                new_value=None,
                version=current.version,
                timestamp=time.time(),
                operation=OperationType.DELETE
            ))
        
        return success
    
    # ==================== 观察者模式 ====================
    
    def watch(self, key: str, callback: Callable[[StateChange], None]) -> str:
        """注册观察者"""
        watcher_id = str(uuid.uuid4())
        if key not in self._watchers:
            self._watchers[key] = []
        self._watchers[key].append(callback)
        return watcher_id
    
    def unwatch(self, key: str, watcher_id: str) -> None:
        """取消观察者"""
        if key in self._watchers:
            self._watchers[key] = [
                w for w in self._watchers[key] 
                if getattr(w, '__id__', None) != watcher_id
            ]
    
    async def _notify_watchers(self, change: StateChange) -> None:
        """通知所有观察者"""
        if change.key in self._watchers:
            for callback in self._watchers[change.key]:
                try:
                    if asyncio.iscoroutinefunction(callback):
                        await callback(change)
                    else:
                        callback(change)
                except Exception as e:
                    print(f"Watcher callback error: {e}")
    
    # ==================== 批量操作 ====================
    
    async def batch_get(self, keys: List[str]) -> Dict[str, Optional[StateEntry]]:
        """批量获取"""
        results = {}
        for key in keys:
            results[key] = await self.get(key)
        return results
    
    async def batch_set(
        self, 
        items: Dict[str, Any],
        metadata: Dict[str, Any] = None
    ) -> List[StateEntry]:
        """批量设置"""
        results = []
        for key, value in items.items():
            entry = await self.set(key, value, metadata=metadata)
            results.append(entry)
        return results
    
    async def scan(self, pattern: str) -> List[StateEntry]:
        """扫描匹配的模式"""
        return await self.storage.scan(pattern)
    
    # ==================== 指标 ====================
    
    def get_metrics(self) -> Dict[str, int]:
        """获取指标"""
        return self._metrics.copy()

# example_usage.py
import asyncio
from state_manager.storage import RedisStorage, MemoryStorage
from state_manager.manager import StateManager, OptimisticLock

async def main():
    # 1. 创建存储后端（使用内存存储演示）
    storage = MemoryStorage()
    
    # 2. 创建状态管理器
    lock_config = OptimisticLock(
        enabled=True,
        max_retries=5,
        retry_delay=0.1
    )
    manager = StateManager(storage, lock_config)
    
    # 3. 基本CRUD操作
    print("=== 基础操作 ===")
    
    # 创建
    task1 = await manager.set(
        key="task:001",
        value={
            "title": "生成代码",
            "status": "pending",
            "progress": 0.0
        }
    )
    print(f"Created task: {task1.key}, version={task1.version}")
    
    # 读取
    task1_fetched = await manager.get("task:001")
    print(f"Fetched task: {task1_fetched.value}")
    
    # 更新
    updated_task = await manager.set(
        key="task:001",
        value={
            "title": "生成代码",
            "status": "running",
            "progress": 0.5
        },
        version=task1.version  # 乐观锁版本
    )
    print(f"Updated task: version={updated_task.version}")
    
    # 4. 乐观锁冲突演示
    print("\n=== 乐观锁冲突 ===")
    
    # 模拟两个并发请求
    async def update_from_user_a():
        try:
            # 用户A读取（version=2）
            entry = await manager.get("task:001")
            # 模拟用户A的处理时间
            await asyncio.sleep(0.5)
            # 用户A尝试更新
            await manager.set(
                "task:001",
                {"title": "生成代码", "status": "completed", "progress": 1.0},
                version=entry.version
            )
            print("User A: Update succeeded")
        except Exception as e:
            print(f"User A: {e}")
    
    async def update_from_user_b():
        try:
            # 用户B读取（version=2）
            entry = await manager.get("task:001")
            # 用户B立即更新
            await asyncio.sleep(0.1)
            await manager.set(
                "task:001",
                {"title": "生成代码", "status": "failed", "progress": 0.0, "error": "Timeout"},
                version=entry.version
            )
            print("User B: Update succeeded")
        except Exception as e:
            print(f"User B: {e}")
    
    # 并发执行
    await asyncio.gather(
        update_from_user_a(),
        update_from_user_b()
    )
    
    # 5. 观察者模式
    print("\n=== 观察者模式 ===")
    
    async def on_task_change(change):
        print(f"Task changed: {change.key}")
        print(f"  Old: {change.old_value}")
        print(f"  New: {change.new_value}")
        print(f"  Version: {change.version}")
    
    watcher_id = manager.watch("task:001", on_task_change)
    
    await manager.set(
        "task:001",
        {"title": "生成代码", "status": "running", "progress": 0.75},
        version=3
    )
    
    # 6. 批量操作
    print("\n=== 批量操作 ===")
    
    await manager.batch_set({
        "session:user:001": {"user_id": "001", "name": "Alice"},
        "session:user:002": {"user_id": "002", "name": "Bob"},
        "session:user:003": {"user_id": "003", "name": "Charlie"},
    })
    
    sessions = await manager.scan("session:user:*")
    for session in sessions:
        print(f"  {session.key}: {session.value}")
    
    # 7. 函数式更新
    print("\n=== 函数式更新 ===")
    
    await manager.set(
        "counter:tasks",
        0,
        metadata={"purpose": "task counter"}
    )
    
    for i in range(5):
        await manager.update(
            "counter:tasks",
            lambda n: n + 1
        )
    
    counter = await manager.get("counter:tasks")
    print(f"Counter value: {counter.value}")
    
    # 8. 指标
    print("\n=== 指标 ===")
    metrics = manager.get_metrics()
    for key, value in metrics.items():
        print(f"  {key}: {value}")

if __name__ == "__main__":
    asyncio.run(main())

# test_state_manager.py
import pytest
import asyncio
from state_manager.storage import MemoryStorage
from state_manager.manager import StateManager, ConflictError

@pytest.fixture
def manager():
    storage = MemoryStorage()
    return StateManager(storage)

class TestStateManager:
    
    @pytest.mark.asyncio
    async def test_basic_crud(self, manager):
        """测试基础CRUD"""
        # Create
        entry = await manager.set("key1", "value1")
        assert entry.key == "key1"
        assert entry.value == "value1"
        assert entry.version == 1
        
        # Read
        fetched = await manager.get("key1")
        assert fetched.value == "value1"
        
        # Update
        updated = await manager.set("key1", "value2", version=1)
        assert updated.value == "value2"
        assert updated.version == 2
        
        # Delete
        deleted = await manager.delete("key1")
        assert deleted == True
        
        # Not found
        result = await manager.get("key1")
        assert result is None
    
    @pytest.mark.asyncio
    async def test_optimistic_lock_conflict(self, manager):
        """测试乐观锁冲突"""
        # 创建初始值
        entry = await manager.set("key1", "value1")
        assert entry.version == 1
        
        # 使用旧版本更新 - 应该成功（无版本检查）
        await manager.set("key1", "value2", version=1)
        
        # 使用过期版本更新 - 应该失败
        with pytest.raises(ConflictError) as exc_info:
            await manager.set("key1", "value3", version=1)
        
        assert exc_info.value.expected_version == 1
        assert exc_info.value.actual_version == 2
    
    @pytest.mark.asyncio
    async def test_concurrent_updates(self, manager):
        """测试并发更新"""
        await manager.set("counter", 0)
        
        async def increment():
            for _ in range(10):
                try:
                    await manager.update("counter", lambda x: x + 1)
                except ConflictError:
                    pass  # 忽略冲突
        
        # 并发执行
        await asyncio.gather(*[increment() for _ in range(5)])
        
        # 最终值应该是50（如果无冲突）
        # 但由于乐观锁，会有一些冲突和重试
        result = await manager.get("counter")
        assert result.value >= 0  # 至少不会是负数
    
    @pytest.mark.asyncio
    async def test_watcher(self, manager):
        """测试观察者"""
        changes = []
        
        async def watcher(change):
            changes.append(change)
        
        manager.watch("key1", watcher)
        
        await manager.set("key1", "value1")
        await manager.set("key1", "value2")
        await manager.set("key2", "value3")  # 不应该触发
        
        assert len(changes) == 2
        assert changes[0].new_value == "value1"
        assert changes[1].new_value == "value2"
    
    @pytest.mark.asyncio
    async def test_batch_operations(self, manager):
        """测试批量操作"""
        items = {
            "k1": "v1",
            "k2": "v2",
            "k3": "v3"
        }
        
        results = await manager.batch_set(items)
        assert len(results) == 3
        
        fetched = await manager.batch_get(["k1", "k2", "k3"])
        assert len(fetched) == 3
        assert fetched["k1"].value == "v1"
    
    @pytest.mark.asyncio
    async def test_scan(self, manager):
        """测试模式扫描"""
        await manager.batch_set({
            "user:001:session": "s1",
            "user:002:session": "s2",
            "user:001:profile": "p1",
            "task:001": "t1"
        })
        
        user_sessions = await manager.scan("user:*:session")
        assert len(user_sessions) == 2
        
        all_users = await manager.scan("user:*")
        assert len(all_users) == 3
    
    @pytest.mark.asyncio
    async def test_delete_nonexistent(self, manager):
        """测试删除不存在的键"""
        result = await manager.delete("nonexistent")
        assert result == False
    
    @pytest.mark.asyncio
    async def test_version_inheritance(self, manager):
        """测试版本继承"""
        e1 = await manager.set("key1", "v1")
        assert e1.version == 1
        
        e2 = await manager.set("key1", "v2")
        assert e2.version == 2
        
        e3 = await manager.set("key1", "v3", version=e2.version)
        assert e3.version == 3

# requirements.txt
aiohttp>=3.9.0
aioredis>=2.0.1
pytest>=7.4.0
pytest-asyncio>=0.21.0
redis>=5.0.0
etcd3>=0.12.0

# state_manager.yaml
state_manager:
  storage:
    type: redis  # redis, memory
    redis_url: redis://localhost:6379/0
    pool_size: 10
    timeout: 5.0
  
  optimistic_lock:
    enabled: true
    max_retries: 3
    retry_delay: 0.1
  
  watcher:
    enabled: true
    buffer_size: 1000
    flush_interval: 1.0

distributed_lock:
  redis_hosts:
    - localhost:6379
  retry_count: 3
  retry_delay: 0.2
  auto_release_time: 30000  # ms

# benchmark.py
import asyncio
import time
from state_manager.storage import MemoryStorage
from state_manager.manager import StateManager

async def benchmark(manager, num_ops=10000):
    """基准测试"""
    start = time.time()
    
    # 写入测试
    for i in range(num_ops):
        await manager.set(f"bench:{i}", {"data": i})
    
    write_time = time.time() - start
    
    # 读取测试
    start = time.time()
    for i in range(num_ops):
        await manager.get(f"bench:{i}")
    read_time = time.time() - start
    
    print(f"Write: {num_ops/write_time:.2f} ops/sec")
    print(f"Read: {num_ops/read_time:.2f} ops/sec")

if __name__ == "__main__":
    storage = MemoryStorage()
    manager = StateManager(storage)
    asyncio.run(benchmark(manager))