# photon-etcd-cluster

@README.md

## Dev Commands

```bash
# Build
cargo build

# Test (integration tests need Docker for etcd)
cargo test -- --test-threads=1
```

## Key Files

| File | Purpose |
|------|---------|
| `src/cluster_node/mod.rs` | Leader election + node registration + `ClusterNodeBuilder` |
| `src/cluster_node/metrics_task.rs` | Background task for periodic metrics updates |
| `src/discovery/mod.rs` | Reactive cluster state via events and watch channels |
| `src/metrics.rs` | `MetricsCollector` trait + `SystemMetricsCollector`, `NoopMetricsCollector` |
| `src/event.rs` | `ClusterEvent` enum for state change notifications |
| `src/types.rs` | `Node` struct + `NodeMetadata` + `HealthStatus` enum |
| `src/connection.rs` | etcd connection with exponential backoff |
| `src/error.rs` | `ClusterError` type |

## Architecture

```
ClusterNode (per worker)              ServiceDiscovery (per load-balancer)
├── run() → main loop                 ├── run() → watch loop
├── is_leader() → bool                ├── subscribe() → Receiver<ClusterEvent>
├── current_health() → status         ├── watch_nodes() → Receiver<Arc<Vec<Node>>>
└── resign/leave cleanup              ├── nodes() → Arc<Vec<Node>>
                                      └── leader() → Option<Node>
```

## Patterns

### Reactive API (ServiceDiscovery)
- **Event subscription** via `subscribe()`: Push-based `ClusterEvent` notifications
- **Watch channels** via `watch_nodes()`/`watch_leader()`: Efficient state observation
- **Direct access** via `nodes()`/`leader()`: O(1) lock-free reads via `watch::Receiver::borrow()`

### ClusterEvent types
- `NodeJoined(Node)` / `NodeLeft(Node)` / `NodeUpdated { old, new }`
- `LeaderElected(Node)` / `LeaderLost`
- `Ready` / `Disconnected` / `Reconnected`

### Convenience helpers
- `wait_for_nodes(n)` - wait until n nodes registered
- `wait_for_leader()` - wait until leader elected
- `wait_ready()` - wait for initial etcd sync

### Atomic state
- `ClusterNode.is_leader` uses `AtomicBool` with `Acquire/Release` ordering
- `AtomicHealth` wraps `AtomicU8` for health status

### Graceful shutdown
- All loops accept `broadcast::Receiver<()>` for shutdown signal
- `tokio::select!` used for cancellation points
- ClusterNode resigns leadership and revokes lease on shutdown

### etcd key structure
```
registry/{group_name}/{node_id}  →  Node JSON
election/{group_name}            →  leader election key
```

## Gotchas

- Integration tests require Docker (spawns etcd container on port 2377)
- Run tests with `--test-threads=1` to avoid etcd port conflicts
- Error type is `ClusterError` in `error.rs`

## Code Style

- Use `thiserror` for error types
- Use `tracing` macros (`info!`, `warn!`, `error!`, `debug!`)
- Prefer `Arc<AtomicBool>` over `Arc<Mutex<bool>>` for simple flags
- Use `tokio::sync::watch` for shared state that changes infrequently
- Avoid `.unwrap()` - propagate errors or use `?`

## Testing

- Unit tests: inline `#[cfg(test)]` modules
- Integration tests: `tests/integration.rs` + `tests/common.rs`
- Performance test in `discovery.rs`: validates 10M accesses < 1s

## Dependencies

| Crate | Usage |
|-------|-------|
| `etcd-client` | etcd v3 API (campaign, proclaim, watch, lease) |
| `tokio` | Async runtime, watch/broadcast channels, timers |
| `serde` / `serde_json` | Node serialization |
| `tracing` | Structured logging |
| `thiserror` | Error derive macro |

## API Quick Reference

```rust
// ClusterNode (worker process) - basic
let node = ClusterNode::new(endpoints, id, ip, group, Some(ttl));
node.run(&mut shutdown_rx).await?;
node.is_leader()          // → bool
node.current_health()     // → HealthStatus

// ClusterNode with metrics (using builder)
let node = ClusterNodeBuilder::new(endpoints, id, ip, group)
    .ttl(5)
    .metrics_collector(SystemMetricsCollector::new())  // requires "system-metrics" feature
    .metrics_update_interval(5)  // update every 5 seconds
    .build();

// ServiceDiscovery (load-balancer)
let discovery = ServiceDiscovery::new(endpoints, group);
tokio::spawn(async move { discovery.run(None).await });

// Event-driven (recommended)
let mut events = discovery.subscribe();
while let Ok(event) = events.recv().await {
    match event {
        ClusterEvent::NodeJoined(n) => { /* add backend */ }
        ClusterEvent::NodeLeft(n) => { /* remove backend */ }
        ClusterEvent::NodeUpdated { old, new } => { /* metadata or IP changed */ }
        _ => {}
    }
}

// Watch-based (for metrics)
let mut watch = discovery.watch_nodes();
while watch.changed().await.is_ok() {
    NODE_COUNT.set(watch.borrow().len() as i64);
}

// Direct access
discovery.nodes()              // → Arc<Vec<Node>>
discovery.leader()             // → Option<Node>

// Node metadata access
for node in discovery.nodes().iter() {
    let cpu = node.metadata["cpu_usage_percent"].as_f64();
    let mem = node.metadata["memory_usage_percent"].as_f64();
}

// Convenience helpers
discovery.wait_for_nodes(1).await;  // wait for 1+ nodes
discovery.wait_for_leader().await;  // wait for leader
discovery.wait_ready().await;       // wait for initial sync
```

## Node Metadata (Schema-less)

Nodes can report arbitrary JSON metadata (CPU, memory, custom metrics):

```rust
// Common metadata keys (see metadata_keys module)
metadata_keys::CPU_USAGE_PERCENT      // "cpu_usage_percent"
metadata_keys::MEMORY_USAGE_PERCENT   // "memory_usage_percent"
metadata_keys::MEMORY_AVAILABLE_BYTES // "memory_available_bytes"
metadata_keys::LOAD_AVG_1M            // "load_avg_1m"
metadata_keys::ACTIVE_CONNECTIONS     // "active_connections"
metadata_keys::REQUESTS_PER_SECOND    // "requests_per_second"
metadata_keys::QUEUE_DEPTH            // "queue_depth"
```

### Custom Metrics Collector

```rust
use photon_etcd_cluster::{MetricsCollector, NodeMetadata};
use serde_json::json;

struct MyAppMetrics { /* ... */ }

impl MetricsCollector for MyAppMetrics {
    fn collect(&self) -> NodeMetadata {
        json!({
            "cpu_usage_percent": 45.0,
            "queue_depth": 10,
            "custom_metric": "value"
        })
    }
}

let node = ClusterNodeBuilder::new(endpoints, id, ip, group)
    .metrics_collector(MyAppMetrics::new())
    .metrics_update_interval(5)
    .build();
```

### Composite Metrics (combine sources)

```rust
let collector = CompositeMetricsCollector::new()
    .with_collector(SystemMetricsCollector::new())
    .with_collector(MyAppMetrics::new());  // App metrics override/extend system metrics
```
