# Distributed State Management

State management across multiple nodes: consensus, replication, partitioning, and consistency.

---

## Challenges

| Challenge | Centralized | Distributed |
|---|---|---|
| **Consistency** | Easy (single source) | Hard (CAP theorem) |
| **Availability** | Single point of failure | High (no single point) |
| **Latency** | Low (local) | Higher (network) |
| **Scaling** | Vertical | Horizontal |
| **Debugging** | Simple | Complex |

---

## Consistency Models

### CAP Theorem Applied
```
For Agent State:
├── CP (Consistency + Partition Tolerance)
│   └── Financial, legal, safety-critical agents
├── AP (Availability + Partition Tolerance)  
│   └── Chat, creative, research agents
└── CA (Consistency + Availability)
    └── Single datacenter, no partitions
```

### Consistency Levels
| Level | Guarantees | Latency | Use Case |
|---|---|---|---|
| **Strong (Linearizable)** | Read sees latest write | Highest | Financial, legal |
| **Sequential** | Operations appear sequential | High | Multi-step workflows |
| **Causal** | Cause before effect | Medium | Conversational |
| **Eventual** | Eventually consistent | Lowest | Chat, creative |
| **Read Your Writes** | User sees own writes | Low | User-facing |

---

## State Partitioning (Sharding)

### Partition Strategies
```python
class StatePartitioner:
    def __init__(self, num_shards: int):
        self.num_shards = num_shards
    
    def get_shard(self, key: str) -> int:
        # Consistent hashing
        return hash(key) % self.num_shards
    
    # Partition by:
    # - Thread ID (conversation isolation)
    # - User ID (user data locality)
    # - Tenant ID (multi-tenant isolation)
    # - Agent ID (agent-specific state)
```

### Shard Management
```python
class ShardManager:
    def __init__(self, shards: list[Shard]):
        self.shards = {s.shard_id: s for s in shards}
    
    async def get_shard_for_thread(self, thread_id: str) -> Shard:
        shard_id = self.partitioner.get_shard(thread_id)
        return self.shards[shard_id]
    
    async def rebalance(self, new_num_shards: int):
        # 1. Compute new assignments
        # 2. Migrate data (online or offline)
        # 3. Update routing
        # 4. Verify consistency
```

---

## Distributed Consensus

### Raft for State Machine
```python
# State machine replication via Raft
class RaftStateMachine:
    def __init__(self, node_id: str, peers: list[str]):
        self.node_id = node_id
        self.peers = peers
        self.log: list[LogEntry] = []
        self.state = "follower"
        self.current_term = 0
        self.voted_for = None
    
    async def propose(self, command: Command) -> Result:
        if self.state != "leader":
            raise NotLeader()
        
        entry = LogEntry(term=self.current_term, command=command)
        self.log.append(entry)
        
        # Replicate to quorum
        await self.replicate_to_quorum(entry)
        
        # Apply to state machine
        self.apply(entry.command)
        return Result(success=True)
```

### When to Use Consensus
| Scenario | Consensus Needed? |
|---|---|
| Single-writer (per thread) | No (single leader) |
| Multi-writer same key | Yes |
| Configuration changes | Yes |
| Leader election | Yes |
| Idempotent operations | No (retry + deduplicate) |

---

## Distributed Locking

### Redlock (Redis)
```python
class DistributedLock:
    def __init__(self, redis_clients: list[redis.Redis]):
        self.clients = redis_clients
        self.quorum = len(redis_clients) // 2 + 1
    
    async def acquire(self, resource: str, ttl: int = 10) -> bool:
        lock_id = str(uuid4())
        start = time.time()
        
        acquired = 0
        for client in self.clients:
            if await client.set(f"lock:{resource}", lock_id, nx=True, ex=ttl):
                acquired += 1
        
        if acquired >= self.quorum:
            return True
        
        # Failed: release any acquired
        for client in self.clients:
            await client.eval("if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end", 1, f"lock:{resource}", lock_id)
        return False
    
    async def release(self, resource: str, lock_id: str):
        for client in self.clients:
            await client.eval("if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end", 1, f"lock:{resource}", lock_id)
```

### When to Lock
| Scenario | Lock Needed? |
|---|---|
| Single-threaded per thread | No |
| Multi-agent same resource | Yes |
| Idempotent operations | No |
| External API (non-idempotent) | Yes |

---

## Distributed Caching

### Cache Patterns
```python
class DistributedCache:
    def __init__(self, cache: CacheBackend, local_cache: LocalCache):
        self.remote = cache
        self.local = local_cache  # L1 cache
    
    async def get(self, key: str):
        # L1 check
        value = await self.local.get(key)
        if value is not None:
            return value
        
        # L2 check
        value = await self.remote.get(key)
        if value is not None:
            await self.local.set(key, value)
            return value
        return None
    
    async def set(self, key: str, value, ttl: int):
        await self.remote.set(key, value, ttl)
        await self.local.set(key, value, min(ttl, 300))  # L1 shorter TTL
```

### Cache Invalidation
```python
class CacheInvalidator:
    def __init__(self, cache: DistributedCache, pubsub: PubSub):
        self.cache = cache
        self.pubsub = pubsub
        await self.pubsub.subscribe("cache:invalidate", self.on_invalidate)
    
    async def on_invalidate(self, message):
        keys = message.data["keys"]
        for key in keys:
            await self.cache.local.delete(key)
            await self.cache.remote.delete(key)
    
    async def invalidate(self, keys: list[str]):
        await self.pubsub.publish("cache:invalidate", {"keys": keys})
```

---

## Distributed Sessions

### Session Affinity
```python
class SessionRouter:
    def __init__(self, instances: list[Instance]):
        self.instances = instances
    
    def route(self, session_id: str) -> Instance:
        # Consistent hashing for session affinity
        return self.instances[hash(session_id) % len(self.instances)]
```

### Session Migration
```python
class SessionMigrator:
    async def migrate(self, session_id: str, from_node: Node, to_node: Node):
        # 1. Pause session on source
        await from_node.pause_session(session_id)
        
        # 2. Export state
        state = await from_node.export_session(session_id)
        
        # 3. Import to target
        await to_node.import_session(session_id, state)
        
        # 3. Resume on target
        await to_node.resume_session(session_id)
        
        # 4. Update routing
        await self.update_routing(session_id, to_node)
```

---

## Eventual Consistency Patterns

### Conflict Resolution
```python
class ConflictResolver:
    def resolve(self, local: State, remote: State, strategy: str) -> State:
        if strategy == "last_write_wins":
            return local if local.updated_at > remote.updated_at else remote
        
        elif strategy == "merge":
            return self.merge_states(local, remote)
        
        elif strategy == "application_specific":
            return self.custom_merge(local, remote)
        
        elif strategy == "user_resolution":
            raise ConflictError(local, remote)
    
    def merge_states(self, local: State, remote: State) -> State:
        # CRDT-style merge for compatible types
        merged = {}
        for key in set(local.keys()) | set(remote.keys()):
            l, r = local.get(key), remote.get(key)
            if isinstance(l, dict) and isinstance(r, dict):
                merged[key] = self.merge_states(l, r)
            elif isinstance(l, list) and isinstance(r, list):
                merged[key] = list(set(l + r))  # Union
            elif isinstance(l, (int, float)) and isinstance(r, (int, float)):
                merged[key] = max(l, r)  # Or sum, depending on semantics
            else:
                merged[key] = r if r.timestamp > l.timestamp else l
        return merged
```

### CRDTs (Conflict-free Replicated Data Types)
```python
# G-Counter (Grow-only Counter)
class GCounter:
    def __init__(self, node_id: str):
        self.counts = defaultdict(int)
        self.node_id = node_id
    
    def increment(self):
        self.counts[self.node_id] += 1
    
    def merge(self, other: 'GCounter'):
        for node, count in other.counts.items():
            self.counts[node] = max(self.counts[node], count)
    
    def value(self) -> int:
        return sum(self.counts.values())

# LWW-Register (Last-Writer-Wins)
class LWWRegister:
    def __init__(self):
        self.value = None
        self.timestamp = 0
        self.node_id = None
    
    def set(self, value, timestamp, node_id):
        if timestamp > self.timestamp:
            self.value = value
            self.timestamp = timestamp
            self.node_id = node_id
    
    def merge(self, other: 'LWWRegister'):
        if other.timestamp > self.timestamp:
            self.value = other.value
            self.timestamp = other.timestamp
            self.node_id = other.node_id
```

---

## Multi-Region Deployment

### Active-Passive
```
Region A (Primary) ←→ Region B (Standby)
     │                      │
     ▼                      ▼
  Writes                  Reads
  Sync                    Async
```

### Active-Active
```
Region A ←→ Region B ←→ Region C
  │           │           │
  ▼           ▼           ▼
All Read/Write (Conflict Resolution Required)
```

### Data Residency
```python
class ResidencyRouter:
    def __init__(self, regions: dict[str, Region]):
        self.regions = regions
    
    def route(self, user_id: str, data_type: str) -> Region:
        # Legal/compliance routing
        residency = self.get_residency_requirement(user_id, data_type)
        if residency:
            return self.regions[residency]
        
        # Latency-based
        return self.nearest_region(user_id)
```

---

## Monitoring Distributed State

### Key Metrics
| Metric | Alert Threshold |
|---|---|
| **Replication Lag** | > 1s (sync), > 30s (async) |
| **Consistency Violations** | > 0 |
| **Split Brain Events** | > 0 |
| **Lock Contention** | > 10% wait time |
| **Cache Hit Rate** | < 80% |
| **Cross-Region Latency** | > 100ms (p99) |

---

## Related Notes

- [[StateManagement/Checkpointing.md]] — Distributed checkpointing
- [[StateManagement/Event-Sourcing.md]] — Distributed event sourcing
- [[StateManagement/Session-Management.md]] — Distributed sessions
- [[Production/Scaling-Strategies.md]] — Horizontal scaling
- [[Production/Reliability.md]] — Circuit breakers, failover
- [[Production/Deployment-Architecture.md]] — Multi-region
- [[Production/Reliability.md]] — Circuit breakers
- [[Frontiers/Research-Directions.md]] — CRDTs, consensus research

---

*Sources: "Designing Data-Intensive Applications" (Kleppmann), "Distributed Systems" (Coulouris et al.), Raft paper, Redis Redlock, CRDT papers (Shapiro et al.), DynamoDB, Cassandra, CockroachDB, Spanner, etcd, Consul, distributed systems patterns.*