# Plan: Add Coordinator Microservice

## Context
NowChess scales `core` horizontally via shared Redis but lacks:
- **Instance visibility**: no way to list running cores or their load
- **Load balancing**: games land randomly on cores; no rebalancing
- **Failover**: dead cores orphan subscriptions; bullet chess requires <1s recovery
- **Auto-scaling**: manual ops to add/remove cores
- **Cache management**: no eviction of stale games from core memory

Bullet chess games run on move timings of <3s. 30s failover = game lost on clock. Target: **<300ms failover**.

---

## Architecture: Sub-1s Failover

### Why Not Polling/TTL
- TTL expiry: minimum 10-30s detection
- HTTP polling 3x failure: 30s minimum
- **gRPC streaming TCP drop: 50-200ms** — use this as primary

### Primary: gRPC Bidirectional Streaming
- Core opens a **persistent bidirectional stream** (`CoreHeartbeatStream`) to coordinator on startup
- Core sends heartbeat frames every **200ms**
- Core crash = TCP RST/FIN → coordinator stream error in **~50-200ms**
- Stream also carries metadata updates (subscription count changes) in real-time

### Fallback: Redis Heartbeat + K8s Watch
- Redis heartbeat key `{prefix}:instances:{instanceId}` with **5s TTL**, refreshed every **2s**
- K8s pod watch via Kubernetes Java client (event-driven; handles pod eviction/OOMKill)
- Fallback covers: network partition (TCP stays up but core is zombie), coordinator restart gap

---

## Design

### 1. Module: `modules/coordinator`
**Language**: Scala 3.5.1, Quarkus REST + gRPC  
**Ports**: HTTP 8086, gRPC 9086  
**Dependencies**: Redisson, Kubernetes Java client, Quarkus gRPC  
**Persistence**: None (all state in Redis)

---

### 2. Instance Registry

**Redis schema**:
```
{prefix}:instances:{instanceId}
  - TTL: 5s (refreshed by core every 2s via background task)
  - Value: JSON
    {
      "instanceId": "core-abc123",
      "hostname": "core-pod-3",
      "httpPort": 8080,
      "grpcPort": 9080,
      "subscriptionCount": 147,
      "localCacheSize": 147,
      "lastHeartbeat": "2026-04-26T10:15:30.123Z"
    }

{prefix}:instance:{instanceId}:games
  - Type: Redis Set (no TTL — managed explicitly)
  - Members: all gameIds currently subscribed on this instance
```

**Core changes** (new `InstanceHeartbeatService` bean in `modules/core`):
- `@PostConstruct`: generate stable `instanceId` (hostname + random suffix); open gRPC stream to coordinator; publish Redis heartbeat; register in `{prefix}:instances:{instanceId}`
- Every 200ms: send heartbeat frame on gRPC stream (carries `subscriptionCount`)
- Every 2s: refresh Redis heartbeat bucket TTL
- `subscribeGame(gameId)`: `SADD {prefix}:instance:{instanceId}:games gameId`
- `unsubscribeGame(gameId)` / `evictGame(gameId)`: `SREM {prefix}:instance:{instanceId}:games gameId`
- `@PreDestroy`: delete Redis key + games set; close gRPC stream (clean shutdown)

---

### 3. Health Monitoring (3 signals, primary fast)

| Signal | Mechanism | Detection time | Role |
|--------|-----------|---------------|------|
| **gRPC stream drop** | TCP RST/FIN on bidirectional stream | 50–200ms | Primary |
| **Redis heartbeat expiry** | `{prefix}:instances:{instanceId}` TTL=5s | 5–7s | Fallback |
| **K8s pod watch** | `CoreV1Api.listNamespacedPod` watch stream | ~instant (pod events) | Fallback |

**Dead decision**:
- gRPC stream drops → **immediate failover** (no confirmation needed; games must recover fast)
- Redis heartbeat expires (gRPC still up) → verify with single HTTP `/q/health` call → if fail: failover
- K8s pod NotReady (gRPC still up) → failover

---

### 4. Failover Protocol (<300ms target)

```
T+0ms     Core JVM crashes / network drops
T+50ms    Coordinator: gRPC stream error received
T+52ms    SMEMBERS {prefix}:instance:{instanceId}:games  → list of orphaned gameIds
T+55ms    Distribute gameIds across healthy cores (least-loaded first)
T+60ms    BatchResubscribeGames gRPC call(s) fire to healthy core(s)
T+150ms   Healthy cores resubscribed; Redis s2c topics live again
T+200ms   WebSocket clients reconnect; receive GameFullEventDto on CONNECTED
```

**Failover steps** (coordinator `FailoverService`):
1. On stream drop for `instanceId`:
   a. Mark instance DEAD in local map
   b. `SMEMBERS {prefix}:instance:{instanceId}:games`
   c. Group gameIds into batches per target core (round-robin by load)
   d. For each target core: call `BatchResubscribeGames(gameIds)`
   e. Each target core: calls `subscribeGame(gameId)` for each (loads from Redis if not in local cache)
   f. `DEL {prefix}:instance:{instanceId}:games` (cleanup)
2. Log failover event with count of games migrated + latency

---

### 5. Load Rebalancing

**Thresholds** (both must be evaluated):
1. **Absolute**: any core > 500 games → rebalance
2. **Relative**: max load > mean × 1.2 AND max - min > 50 games → rebalance

**Algorithm** (runs every 30s, min 60s between actual rebalances):
1. Read all `{prefix}:instances:*` keys → load map
2. Identify overloaded cores (exceed either threshold)
3. For each overloaded core: pick `excess = load - targetLoad` games
4. Assign excess games to underloaded cores
5. Call `UnsubscribeGames(gameIds)` on overloaded core
6. Call `BatchResubscribeGames(gameIds)` on target core
7. Overloaded core: `SREM` each game from its set
8. Target core: `SADD` each game to its set on subscribe

---

### 6. Auto-Scaling

**Metric**: avg `subscriptionCount` across all cores

**Actions**:
- avg > `scale-up-threshold` (80% of max): patch `nowchess-core` Argo Rollout `spec.replicas += 1`
- avg < `scale-down-threshold` (30% of max) AND `replicas > min-replicas`: drain one core then scale down
- Backoff: min 2-minute interval between scale events

**Argo Rollouts API**:
- CRD: `argoproj.io/v1alpha1`, Kind: `Rollout`, resource: `rollouts`
- Scale via Fabric8 `GenericKubernetesResource` patch on `spec.replicas`
- No StatefulSet — Argo Rollout owns pod lifecycle (canary/blue-green strategies respected)
- Pod watch filter: label selector `app=nowchess-core` (Rollout sets this; `rollouts-pod-template-hash` is Argo's equivalent of `pod-template-hash`)

**Drain before scale-down**:
1. Pick least-loaded core
2. Migrate all its games to other cores via `BatchResubscribeGames`
3. Call `DrainInstance(instanceId)` on that core (sets it to reject new subscriptions)
4. After drain confirmed: patch Rollout `spec.replicas -= 1`

---

### 7. Cache Eviction

**Trigger**: coordinator scans `{prefix}:game:entry:*` every 10 minutes  
**Policy**: if `now - lastUpdated > 45min` AND `gameId` in any instance's games set → call `EvictGame`  
**Effect**: core removes game from `localEngines` and `unsubscribeGame`, `SREM` from instance set

---

### 8. Proto: `coordinator_service.proto`

```proto
syntax = "proto3";
package de.nowchess.coordinator;

service CoordinatorService {
  // Core → Coordinator: bidirectional stream for liveness
  rpc HeartbeatStream(stream HeartbeatFrame) returns (stream CoordinatorCommand);

  // Coordinator → Core: batch resubscribe after failover or rebalance
  rpc BatchResubscribeGames(BatchResubscribeRequest) returns (BatchResubscribeResponse);

  // Coordinator → Core: unsubscribe games (rebalance source)
  rpc UnsubscribeGames(UnsubscribeGamesRequest) returns (UnsubscribeGamesResponse);

  // Coordinator → Core: evict idle games from local cache
  rpc EvictGames(EvictGamesRequest) returns (EvictGamesResponse);

  // Coordinator → Core: drain instance before scale-down
  rpc DrainInstance(DrainInstanceRequest) returns (DrainInstanceResponse);
}

message HeartbeatFrame {
  string instanceId = 1;
  string hostname = 2;
  int32 httpPort = 3;
  int32 grpcPort = 4;
  int32 subscriptionCount = 5;
  int32 localCacheSize = 6;
  int64 timestampMillis = 7;
}

message CoordinatorCommand {
  // Future: coordinator can push commands back (e.g., "start draining")
  string type = 1;
  string payload = 2;
}

message BatchResubscribeRequest {
  repeated string gameIds = 1;
}

message BatchResubscribeResponse {
  int32 subscribedCount = 1;
  repeated string failedGameIds = 2;
}

message UnsubscribeGamesRequest {
  repeated string gameIds = 1;
}

message UnsubscribeGamesResponse {
  int32 unsubscribedCount = 1;
}

message EvictGamesRequest {
  repeated string gameIds = 1;
}

message EvictGamesResponse {
  int32 evictedCount = 1;
}

message DrainInstanceRequest {}

message DrainInstanceResponse {
  int32 gamesMigrated = 0;
}
```

---

### 9. Coordinator REST API (internal)

- `GET /api/coordinator/instances` — all cores with load, health state
- `GET /api/coordinator/metrics` — load distribution, rebalance history
- `POST /api/coordinator/rebalance` — manual rebalance trigger
- `POST /api/coordinator/failover/{instanceId}` — manual failover
- `POST /api/coordinator/scale-up` / `scale-down` — manual scaling

---

### 10. Configuration

**`modules/coordinator/src/main/resources/application.yml`**:
```yaml
quarkus.application.name: nowchess-coordinator
quarkus.http.port: 8086
quarkus.grpc.server.port: 9086

nowchess.coordinator.max-games-per-core: 500
nowchess.coordinator.max-deviation-percent: 20
nowchess.coordinator.rebalance-interval: 30s
nowchess.coordinator.rebalance-min-interval: 60s
nowchess.coordinator.heartbeat-ttl: 5s
nowchess.coordinator.stream-heartbeat-interval: 200ms
nowchess.coordinator.cache-eviction-interval: 10m
nowchess.coordinator.game-idle-threshold: 45m
nowchess.coordinator.auto-scale-enabled: false
nowchess.coordinator.scale-up-threshold: 0.8
nowchess.coordinator.scale-down-threshold: 0.3
nowchess.coordinator.scale-min-replicas: 2
nowchess.coordinator.scale-max-replicas: 10
nowchess.coordinator.k8s-namespace: default
nowchess.coordinator.k8s-rollout-name: nowchess-core
nowchess.coordinator.k8s-rollout-label-selector: app=nowchess-core

quarkus.kubernetes-client.trust-certs: true
```

**Core `application.yml` additions**:
```yaml
nowchess.coordinator.host: localhost
nowchess.coordinator.grpc-port: 9086
nowchess.coordinator.stream-heartbeat-interval: 200ms
nowchess.coordinator.redis-heartbeat-interval: 2s
nowchess.coordinator.instance-id: ${HOSTNAME:local}-${quarkus.uuid}
```

---

### 11. Files to Create / Modify

**New — `modules/coordinator/`**:
```
build.gradle.kts
src/main/proto/coordinator_service.proto
src/main/resources/application.yml
src/main/scala/de/nowchess/coordinator/
  resource/CoordinatorResource.scala       # REST endpoints
  service/InstanceRegistry.scala           # Redis instance list + in-memory map
  service/HealthMonitor.scala              # gRPC stream watcher + Redis TTL + k8s watch
  service/FailoverService.scala            # dead core → BatchResubscribe
  service/LoadBalancer.scala               # rebalance logic
  service/AutoScaler.scala                 # k8s StatefulSet scaling
  service/CacheEvictionManager.scala       # idle game eviction
  grpc/CoordinatorGrpcServer.scala         # CoordinatorService gRPC impl (for HeartbeatStream)
```

**Modify — `modules/core/`**:
- `build.gradle.kts` — add `coordinator_service.proto` stub, keep grpc dep
- `src/main/proto/coordinator_service.proto` — copy (or symlink) proto for stub generation
- `src/main/scala/de/nowchess/chess/redis/GameRedisSubscriberManager.scala` — `SADD`/`SREM` on subscribe/unsubscribe + implement `BatchResubscribeGames`, `UnsubscribeGames`, `EvictGames`, `DrainInstance` gRPC handlers
- `src/main/scala/de/nowchess/chess/` — new `InstanceHeartbeatService.scala` (startup, gRPC stream, Redis TTL refresh)
- `src/main/resources/application.yml` — coordinator connection config

**Modify — root**:
- `settings.gradle.kts` — add `include("modules/coordinator")`

---

## Verification

1. `./compile` — coordinator and core compile cleanly
2. **Stream detection**: start core + coordinator; kill core JVM (`kill -9`); coordinator logs failover within 300ms
3. **Game continuity**: active game on killed core; WebSocket client reconnects and receives game state
4. **Rebalance**: create 600 games on core-1 (2-core setup); coordinator rebalances ~100 to core-2
5. **Fallback**: disconnect gRPC stream manually but keep core alive; Redis TTL fallback triggers within 7s
6. **Cache eviction**: create idle game; coordinator calls `EvictGames` after 45min idle
7. **REST metrics**: `curl localhost:8086/api/coordinator/metrics` returns per-core load + health
8. **Restart recovery**: restart coordinator; gRPC streams re-establish from cores; state rebuilt from Redis

---

## Dependencies (new)

- `io.fabric8:kubernetes-client:6.13.0` (Fabric8 k8s client — handles Argo `Rollout` CRD via `GenericKubernetesResource`; no Argo Java SDK needed)
- Redisson — already in core, reuse via shared config
- Quarkus gRPC — already in core, reuse