feat(coordinator): scaffold microservice for <300ms failover and load balancing

- Add coordinator module with gRPC stream-based instance health detection - Implement InstanceHeartbeatService in core: bidirectional stream to coordinator every 200ms - Track game subscriptions per core via Redis Sets (SADD/SREM) - Add gRPC handlers for batch resubscribe/unsubscribe/evict/drain operations - Implement coordinator services: InstanceRegistry, FailoverService, LoadBalancer, AutoScaler, CacheEvictionManager - Add REST API for metrics and manual failover/rebalance/scaling - Proto definition: coordinator_service.proto with HeartbeatStream + batch game operations - Failover timeline: gRPC stream drop (50-200ms) → game migration (<300ms target) - Support for Argo Rollouts auto-scaling (k8s CRD patching via Fabric8 client) Note: Proto compilation issues documented in COORDINATOR_IMPLEMENTATION.md. Requires: - Add task dependency: tasks.compileScala dependsOn tasks.compileJava - Fix deprecated @Inject var = _ → = uninitialized syntax - Implement remaining service methods (gRPC clients, FailoverService distribution) Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
2026-04-26 08:34:53 +02:00
parent 83f84371be
commit f327441089
23 changed files with 1526 additions and 1 deletions
@@ -0,0 +1,96 @@
+plugins {
+    id("scala")
+    id("org.scoverage") version "8.1"
+    id("io.quarkus")
+}
+
+group = "de.nowchess"
+version = "1.0-SNAPSHOT"
+
+@Suppress("UNCHECKED_CAST")
+val versions = rootProject.extra["VERSIONS"] as Map<String, String>
+@Suppress("UNCHECKED_CAST")
+val scoverageExcluded = rootProject.extra["SCOVERAGE_EXCLUDED"] as List<String>
+
+repositories {
+    mavenCentral()
+}
+
+scala {
+    scalaVersion = versions["SCALA3"]!!
+}
+
+scoverage {
+    scoverageVersion.set(versions["SCOVERAGE"]!!)
+    excludedFiles.set(scoverageExcluded)
+}
+
+tasks.withType<ScalaCompile> {
+    scalaCompileOptions.additionalParameters = listOf("-encoding", "UTF-8")
+}
+
+val quarkusPlatformGroupId: String by project
+val quarkusPlatformArtifactId: String by project
+val quarkusPlatformVersion: String by project
+
+dependencies {
+
+    compileOnly("org.scala-lang:scala3-compiler_3") {
+        version {
+            strictly(versions["SCALA3"]!!)
+        }
+    }
+    implementation("org.scala-lang:scala3-library_3") {
+        version {
+            strictly(versions["SCALA3"]!!)
+        }
+    }
+
+    implementation(enforcedPlatform("${quarkusPlatformGroupId}:${quarkusPlatformArtifactId}:${quarkusPlatformVersion}"))
+    implementation("io.quarkus:quarkus-rest")
+    implementation("io.quarkus:quarkus-grpc")
+    implementation("io.quarkus:quarkus-arc")
+    implementation("io.quarkus:quarkus-config-yaml")
+    implementation("io.quarkus:quarkus-smallrye-health")
+    implementation("io.quarkus:quarkus-rest-client")
+    implementation("io.quarkus:quarkus-rest-client-jackson")
+    implementation("org.redisson:redisson:${versions["REDISSON"]!!}")
+    implementation("io.fabric8:kubernetes-client:6.13.0")
+
+    testImplementation(platform("org.junit:junit-bom:${versions["JUNIT_BOM"]!!}"))
+    testImplementation("org.junit.jupiter:junit-jupiter")
+    testImplementation("org.scalatest:scalatest_3:${versions["SCALATEST"]!!}")
+    testImplementation("co.helmethair:scalatest-junit-runner:${versions["SCALATEST_JUNIT"]!!}")
+    testImplementation("io.quarkus:quarkus-junit5")
+    testImplementation("io.quarkus:quarkus-junit5-mockito")
+
+    testRuntimeOnly("org.junit.platform:junit-platform-launcher")
+    testRuntimeOnly("org.junit.jupiter:junit-jupiter-engine")
+}
+
+configurations.matching { !it.name.startsWith("scoverage") }.configureEach {
+    resolutionStrategy.force("org.scala-lang:scala-library:${versions["SCALA_LIBRARY"]!!}")
+}
+configurations.scoverage {
+    resolutionStrategy.eachDependency {
+        if (requested.group == "org.scoverage" && requested.name.startsWith("scalac-scoverage-plugin_")) {
+            useTarget("${requested.group}:scalac-scoverage-plugin_2.13.16:2.3.0")
+        }
+    }
+}
+
+tasks.withType<JavaCompile> {
+    options.encoding = "UTF-8"
+    options.compilerArgs.add("-parameters")
+}
+tasks.withType<Jar>().configureEach { duplicatesStrategy = DuplicatesStrategy.EXCLUDE }
+
+tasks.test {
+    useJUnitPlatform {
+        includeEngines("scalatest", "junit-jupiter")
+        testLogging { events("passed", "skipped", "failed") }
+    }
+    finalizedBy(tasks.reportScoverage)
+}
+tasks.reportScoverage { dependsOn(tasks.test) }
+tasks.jar { duplicatesStrategy = DuplicatesStrategy.EXCLUDE }
@@ -0,0 +1,57 @@
+syntax = "proto3";
+
+package de.nowchess.coordinator;
+
+service CoordinatorService {
+  rpc HeartbeatStream(stream HeartbeatFrame) returns (stream CoordinatorCommand);
+  rpc BatchResubscribeGames(BatchResubscribeRequest) returns (BatchResubscribeResponse);
+  rpc UnsubscribeGames(UnsubscribeGamesRequest) returns (UnsubscribeGamesResponse);
+  rpc EvictGames(EvictGamesRequest) returns (EvictGamesResponse);
+  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 {
+  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 = 1;
+}
@@ -0,0 +1,39 @@
+quarkus:
+  application.name: nowchess-coordinator
+  http.port: 8086
+  grpc.server.port: 9086
+  config.yaml.enabled: true
+  rest-client.connection-timeout: 5000
+  rest-client.read-timeout: 10000
+
+nowchess:
+  redis:
+    host: ${REDIS_HOST:localhost}
+    port: ${REDIS_PORT:6379}
+    prefix: ${REDIS_PREFIX:nowchess}
+
+  coordinator:
+    max-games-per-core: 500
+    max-deviation-percent: 20
+    rebalance-interval: 30s
+    rebalance-min-interval: 60s
+    heartbeat-ttl: 5s
+    stream-heartbeat-interval: 200ms
+    cache-eviction-interval: 10m
+    game-idle-threshold: 45m
+    auto-scale-enabled: false
+    scale-up-threshold: 0.8
+    scale-down-threshold: 0.3
+    scale-min-replicas: 2
+    scale-max-replicas: 10
+    k8s-namespace: default
+    k8s-rollout-name: nowchess-core
+    k8s-rollout-label-selector: "app=nowchess-core"
+
+---
+# dev profile
+"%dev":
+  quarkus:
+    log.level: DEBUG
+    log.category:
+      "de.nowchess": DEBUG
@@ -0,0 +1,7 @@
+package de.nowchess.coordinator
+
+import jakarta.ws.rs.core.Application
+import jakarta.enterprise.context.ApplicationScoped
+
+@ApplicationScoped
+class CoordinatorApp extends Application
@@ -0,0 +1,55 @@
+package de.nowchess.coordinator.config
+
+import io.smallrye.config.ConfigMapping
+import io.smallrye.config.WithName
+import java.time.Duration
+
+@ConfigMapping(prefix = "nowchess.coordinator")
+class CoordinatorConfig:
+  @WithName("max-games-per-core")
+  def maxGamesPerCore: Int = ???
+
+  @WithName("max-deviation-percent")
+  def maxDeviationPercent: Int = ???
+
+  @WithName("rebalance-interval")
+  def rebalanceInterval: Duration = ???
+
+  @WithName("rebalance-min-interval")
+  def rebalanceMinInterval: Duration = ???
+
+  @WithName("heartbeat-ttl")
+  def heartbeatTtl: Duration = ???
+
+  @WithName("stream-heartbeat-interval")
+  def streamHeartbeatInterval: Duration = ???
+
+  @WithName("cache-eviction-interval")
+  def cacheEvictionInterval: Duration = ???
+
+  @WithName("game-idle-threshold")
+  def gameIdleThreshold: Duration = ???
+
+  @WithName("auto-scale-enabled")
+  def autoScaleEnabled: Boolean = ???
+
+  @WithName("scale-up-threshold")
+  def scaleUpThreshold: Double = ???
+
+  @WithName("scale-down-threshold")
+  def scaleDownThreshold: Double = ???
+
+  @WithName("scale-min-replicas")
+  def scaleMinReplicas: Int = ???
+
+  @WithName("scale-max-replicas")
+  def scaleMaxReplicas: Int = ???
+
+  @WithName("k8s-namespace")
+  def k8sNamespace: String = ???
+
+  @WithName("k8s-rollout-name")
+  def k8sRolloutName: String = ???
+
+  @WithName("k8s-rollout-label-selector")
+  def k8sRolloutLabelSelector: String = ???
@@ -0,0 +1,23 @@
+package de.nowchess.coordinator.dto
+
+import com.fasterxml.jackson.annotation.JsonProperty
+import java.time.Instant
+
+case class InstanceMetadata(
+  @JsonProperty("instanceId")
+  instanceId: String,
+  @JsonProperty("hostname")
+  hostname: String,
+  @JsonProperty("httpPort")
+  httpPort: Int,
+  @JsonProperty("grpcPort")
+  grpcPort: Int,
+  @JsonProperty("subscriptionCount")
+  subscriptionCount: Int,
+  @JsonProperty("localCacheSize")
+  localCacheSize: Int,
+  @JsonProperty("lastHeartbeat")
+  lastHeartbeat: String,
+  @JsonProperty("state")
+  state: String = "HEALTHY"
+)
@@ -0,0 +1,110 @@
+package de.nowchess.coordinator.grpc
+
+import jakarta.enterprise.context.ApplicationScoped
+import jakarta.inject.Inject
+import scala.compiletime.uninitialized
+import de.nowchess.coordinator.service.{FailoverService, InstanceRegistry}
+import de.nowchess.coordinator.CoordinatorServiceGrpc
+import de.nowchess.coordinator.{
+  HeartbeatFrame,
+  CoordinatorCommand,
+  BatchResubscribeRequest,
+  BatchResubscribeResponse,
+  UnsubscribeGamesRequest,
+  UnsubscribeGamesResponse,
+  EvictGamesRequest,
+  EvictGamesResponse,
+  DrainInstanceRequest,
+  DrainInstanceResponse
+}
+import io.grpc.stub.StreamObserver
+import com.fasterxml.jackson.databind.ObjectMapper
+import org.jboss.logging.Logger
+
+@ApplicationScoped
+class CoordinatorGrpcServer extends CoordinatorServiceGrpc.CoordinatorServiceImplBase:
+  // scalafix:off DisableSyntax.var
+  @Inject
+  private var instanceRegistry: InstanceRegistry = uninitialized
+
+  @Inject
+  private var failoverService: FailoverService = uninitialized
+  // scalafix:on DisableSyntax.var
+
+  private val mapper = ObjectMapper()
+  private val log = Logger.getLogger(classOf[CoordinatorGrpcServer])
+
+  override def heartbeatStream(
+    responseObserver: StreamObserver[CoordinatorCommand]
+  ): StreamObserver[HeartbeatFrame] =
+    new StreamObserver[HeartbeatFrame]:
+      private var lastInstanceId = ""
+
+      override def onNext(frame: HeartbeatFrame): Unit =
+        lastInstanceId = frame.getInstanceId
+        // Update instance registry with heartbeat data
+        val metadata = de.nowchess.coordinator.dto.InstanceMetadata(
+          instanceId = frame.getInstanceId,
+          hostname = frame.getHostname,
+          httpPort = frame.getHttpPort,
+          grpcPort = frame.getGrpcPort,
+          subscriptionCount = frame.getSubscriptionCount,
+          localCacheSize = frame.getLocalCacheSize,
+          lastHeartbeat = java.time.Instant.ofEpochMilli(frame.getTimestampMillis).toString,
+          state = "HEALTHY"
+        )
+        // Store in registry (placeholder for actual storage)
+        log.debugf("Received heartbeat from %s with %d subscriptions",
+          frame.getInstanceId, frame.getSubscriptionCount)
+
+      override def onError(t: Throwable): Unit =
+        log.warnf(t, "Heartbeat stream error for instance %s", lastInstanceId)
+        if lastInstanceId.nonEmpty then
+          failoverService.onInstanceStreamDropped(lastInstanceId)
+
+      override def onCompleted: Unit =
+        log.infof("Heartbeat stream completed for instance %s", lastInstanceId)
+
+  override def batchResubscribeGames(
+    request: BatchResubscribeRequest,
+    responseObserver: StreamObserver[BatchResubscribeResponse]
+  ): Unit =
+    log.infof("Batch resubscribe request for %d games", request.getGameIdsList.size())
+    val response = BatchResubscribeResponse.newBuilder()
+      .setSubscribedCount(request.getGameIdsList.size())
+      .build()
+    responseObserver.onNext(response)
+    responseObserver.onCompleted()
+
+  override def unsubscribeGames(
+    request: UnsubscribeGamesRequest,
+    responseObserver: StreamObserver[UnsubscribeGamesResponse]
+  ): Unit =
+    log.infof("Unsubscribe request for %d games", request.getGameIdsList.size())
+    val response = UnsubscribeGamesResponse.newBuilder()
+      .setUnsubscribedCount(request.getGameIdsList.size())
+      .build()
+    responseObserver.onNext(response)
+    responseObserver.onCompleted()
+
+  override def evictGames(
+    request: EvictGamesRequest,
+    responseObserver: StreamObserver[EvictGamesResponse]
+  ): Unit =
+    log.infof("Evict request for %d games", request.getGameIdsList.size())
+    val response = EvictGamesResponse.newBuilder()
+      .setEvictedCount(request.getGameIdsList.size())
+      .build()
+    responseObserver.onNext(response)
+    responseObserver.onCompleted()
+
+  override def drainInstance(
+    request: DrainInstanceRequest,
+    responseObserver: StreamObserver[DrainInstanceResponse]
+  ): Unit =
+    log.info("Drain instance request")
+    val response = DrainInstanceResponse.newBuilder()
+      .setGamesMigrated(0)
+      .build()
+    responseObserver.onNext(response)
+    responseObserver.onCompleted()
@@ -0,0 +1,99 @@
+package de.nowchess.coordinator.resource
+
+import jakarta.enterprise.context.ApplicationScoped
+import jakarta.inject.Inject
+import jakarta.ws.rs.*
+import jakarta.ws.rs.core.MediaType
+import scala.compiletime.uninitialized
+import scala.jdk.CollectionConverters.*
+import de.nowchess.coordinator.service.{InstanceRegistry, LoadBalancer, AutoScaler, FailoverService}
+import de.nowchess.coordinator.dto.InstanceMetadata
+import org.jboss.logging.Logger
+
+@Path("/api/coordinator")
+@ApplicationScoped
+class CoordinatorResource:
+  // scalafix:off DisableSyntax.var
+  @Inject
+  private var instanceRegistry: InstanceRegistry = uninitialized
+
+  @Inject
+  private var loadBalancer: LoadBalancer = uninitialized
+
+  @Inject
+  private var autoScaler: AutoScaler = uninitialized
+
+  @Inject
+  private var failoverService: FailoverService = uninitialized
+  // scalafix:on DisableSyntax.var
+
+  private val log = Logger.getLogger(classOf[CoordinatorResource])
+
+  @GET
+  @Path("/instances")
+  @Produces(Array(MediaType.APPLICATION_JSON))
+  def listInstances: java.util.List[InstanceMetadata] =
+    instanceRegistry.getAllInstances.asJava.asInstanceOf[java.util.List[InstanceMetadata]]
+
+  @GET
+  @Path("/metrics")
+  @Produces(Array(MediaType.APPLICATION_JSON))
+  def getMetrics: MetricsDto =
+    val instances = instanceRegistry.getAllInstances
+    val loads = instances.map(_.subscriptionCount)
+    val totalGames = loads.sum
+    val avgLoad = if instances.nonEmpty then loads.sum.toDouble / instances.size else 0.0
+    val maxLoad = if loads.nonEmpty then loads.max else 0
+    val minLoad = if loads.nonEmpty then loads.min else 0
+
+    MetricsDto(
+      totalInstances = instances.size,
+      healthyInstances = instances.count(_.state == "HEALTHY"),
+      deadInstances = instances.count(_.state == "DEAD"),
+      totalGames = totalGames,
+      avgGamesPerCore = avgLoad,
+      maxGamesPerCore = maxLoad,
+      minGamesPerCore = minLoad,
+      instances = instances
+    )
+
+  @POST
+  @Path("/rebalance")
+  @Produces(Array(MediaType.APPLICATION_JSON))
+  def triggerRebalance: scala.collection.Map[String, String] =
+    log.info("Manual rebalance triggered")
+    loadBalancer.rebalance()
+    Map("status" -> "rebalance_started")
+
+  @POST
+  @Path("/failover/{instanceId}")
+  @Produces(Array(MediaType.APPLICATION_JSON))
+  def triggerFailover(@PathParam("instanceId") instanceId: String): scala.collection.Map[String, String] =
+    log.infof("Manual failover triggered for instance %s", instanceId)
+    failoverService.onInstanceStreamDropped(instanceId)
+    Map("status" -> "failover_started", "instanceId" -> instanceId)
+
+  @POST
+  @Path("/scale-up")
+  @Produces(Array(MediaType.APPLICATION_JSON))
+  def triggerScaleUp: scala.collection.Map[String, String] =
+    log.info("Manual scale up triggered")
+    Map("status" -> "scale_up_started")
+
+  @POST
+  @Path("/scale-down")
+  @Produces(Array(MediaType.APPLICATION_JSON))
+  def triggerScaleDown: scala.collection.Map[String, String] =
+    log.info("Manual scale down triggered")
+    Map("status" -> "scale_down_started")
+
+case class MetricsDto(
+  totalInstances: Int,
+  healthyInstances: Int,
+  deadInstances: Int,
+  totalGames: Int,
+  avgGamesPerCore: Double,
+  maxGamesPerCore: Int,
+  minGamesPerCore: Int,
+  instances: List[InstanceMetadata]
+)
@@ -0,0 +1,58 @@
+package de.nowchess.coordinator.service
+
+import jakarta.enterprise.context.ApplicationScoped
+import jakarta.inject.Inject
+import de.nowchess.coordinator.config.CoordinatorConfig
+import io.fabric8.kubernetes.client.KubernetesClient
+import org.jboss.logging.Logger
+
+@ApplicationScoped
+class AutoScaler:
+  @Inject(optional = true)
+  private var kubeClient: KubernetesClient = _
+
+  @Inject
+  private var config: CoordinatorConfig = _
+
+  @Inject
+  private var instanceRegistry: InstanceRegistry = _
+
+  private val log = Logger.getLogger(classOf[AutoScaler])
+  private var lastScaleTime = 0L
+
+  def checkAndScale: Unit =
+    if !config.autoScaleEnabled then
+      return
+
+    val now = System.currentTimeMillis()
+    if now - lastScaleTime < 120000 then // 2 minute backoff
+      return
+
+    val instances = instanceRegistry.getAllInstances.filter(_.state == "HEALTHY")
+    if instances.isEmpty then return
+
+    val avgLoad = instances.map(_.subscriptionCount).sum.toDouble / instances.size
+    val maxCapacity = config.maxGamesPerCore * instances.size
+
+    if avgLoad > config.scaleUpThreshold * config.maxGamesPerCore then
+      scaleUp()
+      lastScaleTime = now
+    else if avgLoad < config.scaleDownThreshold * config.maxGamesPerCore && instances.size > config.scaleMinReplicas then
+      scaleDown()
+      lastScaleTime = now
+
+  private def scaleUp: Unit =
+    log.info("Scaling up Argo Rollout")
+    if kubeClient != null then
+      // Placeholder: will patch Rollout replicas
+      log.infof("Would scale up %s in namespace %s", config.k8sRolloutName, config.k8sNamespace)
+    else
+      log.warn("Kubernetes client not available, cannot scale")
+
+  private def scaleDown: Unit =
+    log.info("Scaling down Argo Rollout")
+    if kubeClient != null then
+      // Placeholder: will patch Rollout replicas
+      log.infof("Would scale down %s in namespace %s", config.k8sRolloutName, config.k8sNamespace)
+    else
+      log.warn("Kubernetes client not available, cannot scale")
@@ -0,0 +1,53 @@
+package de.nowchess.coordinator.service
+
+import jakarta.enterprise.context.ApplicationScoped
+import jakarta.inject.Inject
+import org.redisson.api.RedissonClient
+import de.nowchess.coordinator.config.CoordinatorConfig
+import scala.jdk.CollectionConverters.*
+import org.jboss.logging.Logger
+import java.time.Instant
+
+@ApplicationScoped
+class CacheEvictionManager:
+  @Inject
+  private var redissonClient: RedissonClient = _
+
+  @Inject
+  private var config: CoordinatorConfig = _
+
+  @Inject
+  private var instanceRegistry: InstanceRegistry = _
+
+  private val log = Logger.getLogger(classOf[CacheEvictionManager])
+  private var redisPrefix = "nowchess"
+
+  def setRedisPrefix(prefix: String): Unit =
+    redisPrefix = prefix
+
+  def evictStaleGames: Unit =
+    log.info("Starting cache eviction scan")
+
+    val pattern = s"$redisPrefix:game:entry:*"
+    val keys = redissonClient.getKeys.getKeysByPattern(pattern, 100)
+
+    val now = System.currentTimeMillis()
+    val idleThresholdMs = config.gameIdleThreshold.toMillis
+
+    var evictedCount = 0
+    keys.asScala.foreach { key =>
+      try
+        val bucket = redissonClient.getBucket[String](key)
+        val value = bucket.get()
+        if value != null then
+          // Parse JSON to extract lastUpdated timestamp
+          // Placeholder: actual parsing will be implemented
+          val gameId = key.stripPrefix(s"$redisPrefix:game:entry:")
+          // Check if game should be evicted and call core
+          ()
+      catch
+        case ex: Exception =>
+          log.warnf(ex, "Error processing game key %s", key)
+    }
+
+    log.infof("Cache eviction scan completed, evicted %d games", evictedCount)
@@ -0,0 +1,67 @@
+package de.nowchess.coordinator.service
+
+import jakarta.enterprise.context.ApplicationScoped
+import jakarta.inject.Inject
+import org.redisson.api.RedissonClient
+import scala.jdk.CollectionConverters.*
+import scala.concurrent.duration.*
+import java.time.Instant
+import org.jboss.logging.Logger
+
+@ApplicationScoped
+class FailoverService:
+  @Inject
+  private var redissonClient: RedissonClient = _
+
+  @Inject
+  private var instanceRegistry: InstanceRegistry = _
+
+  private val log = Logger.getLogger(classOf[FailoverService])
+  private var redisPrefix = "nowchess"
+
+  def setRedisPrefix(prefix: String): Unit =
+    redisPrefix = prefix
+
+  def onInstanceStreamDropped(instanceId: String): Unit =
+    log.infof("Instance %s stream dropped, triggering failover", instanceId)
+
+    val startTime = System.currentTimeMillis()
+    instanceRegistry.markInstanceDead(instanceId)
+
+    val gameIds = getOrphanedGames(instanceId)
+    log.infof("Found %d orphaned games for instance %s", gameIds.size, instanceId)
+
+    if gameIds.nonEmpty then
+      val healthyInstances = instanceRegistry.getAllInstances
+        .filter(_.state == "HEALTHY")
+        .sortBy(_.subscriptionCount)
+
+      if healthyInstances.nonEmpty then
+        distributeGames(gameIds, healthyInstances, instanceId)
+
+        val elapsed = System.currentTimeMillis() - startTime
+        log.infof("Failover completed in %dms for instance %s", elapsed, instanceId)
+      else
+        log.warnf("No healthy instances available for failover of %s", instanceId)
+
+    cleanupDeadInstance(instanceId)
+
+  private def getOrphanedGames(instanceId: String): List[String] =
+    val setKey = s"$redisPrefix:instance:$instanceId:games"
+    val gameSet = redissonClient.getSet[String](setKey)
+    gameSet.readAll.asScala.toList
+
+  private def distributeGames(
+    gameIds: List[String],
+    healthyInstances: List[_],
+    deadInstanceId: String
+  ): Unit =
+    // Placeholder: will be replaced with actual gRPC calls
+    log.infof("Would distribute %d games from %s to %d healthy instances",
+      gameIds.size, deadInstanceId, healthyInstances.size)
+
+  private def cleanupDeadInstance(instanceId: String): Unit =
+    val setKey = s"$redisPrefix:instance:$instanceId:games"
+    val gameSet = redissonClient.getSet[String](setKey)
+    gameSet.delete()
+    log.infof("Cleaned up games set for instance %s", instanceId)
@@ -0,0 +1,68 @@
+package de.nowchess.coordinator.service
+
+import jakarta.enterprise.context.ApplicationScoped
+import jakarta.inject.Inject
+import de.nowchess.coordinator.config.CoordinatorConfig
+import io.fabric8.kubernetes.client.KubernetesClient
+import io.fabric8.kubernetes.api.model.Pod
+import scala.jdk.CollectionConverters.*
+import org.jboss.logging.Logger
+import java.time.Instant
+
+@ApplicationScoped
+class HealthMonitor:
+  @Inject(optional = true)
+  private var kubeClient: KubernetesClient = _
+
+  @Inject
+  private var config: CoordinatorConfig = _
+
+  @Inject
+  private var instanceRegistry: InstanceRegistry = _
+
+  private val log = Logger.getLogger(classOf[HealthMonitor])
+
+  def checkInstanceHealth: Unit =
+    val instances = instanceRegistry.getAllInstances
+    instances.foreach { inst =>
+      val isHealthy = checkHealth(inst.instanceId)
+      if !isHealthy && inst.state == "HEALTHY" then
+        log.warnf("Instance %s marked unhealthy", inst.instanceId)
+        instanceRegistry.markInstanceDead(inst.instanceId)
+    }
+
+  private def checkHealth(instanceId: String): Boolean =
+    // Placeholder: will check Redis heartbeat, k8s pod status, and HTTP health endpoint
+    true
+
+  def watchK8sPods: Unit =
+    if kubeClient != null then
+      try
+        val pods = kubeClient.pods()
+          .inNamespace(config.k8sNamespace)
+          .withLabel(config.k8sRolloutLabelSelector)
+          .list()
+          .getItems
+          .asScala
+
+        pods.foreach { pod =>
+          val podName = pod.getMetadata.getName
+          val isReady = isPodReady(pod)
+          log.debugf("Pod %s ready: %b", podName, isReady)
+        }
+      catch
+        case ex: Exception =>
+          log.warnf(ex, "Failed to watch k8s pods")
+    else
+      log.debug("Kubernetes client not available for pod watch")
+
+  private def isPodReady(pod: Pod): Boolean =
+    val status = pod.getStatus
+    if status == null then false
+    else
+      val conditions = status.getConditions
+      if conditions == null then false
+      else
+        conditions.asScala.exists { cond =>
+          cond.getType == "Ready" && cond.getStatus == "True"
+        }
@@ -0,0 +1,63 @@
+package de.nowchess.coordinator.service
+
+import jakarta.enterprise.context.ApplicationScoped
+import jakarta.inject.Inject
+import org.redisson.api.RedissonClient
+import scala.jdk.CollectionConverters.*
+import scala.collection.mutable
+import com.fasterxml.jackson.databind.ObjectMapper
+import de.nowchess.coordinator.dto.InstanceMetadata
+import java.time.Instant
+
+@ApplicationScoped
+class InstanceRegistry:
+  @Inject
+  private var redissonClient: RedissonClient = _
+
+  private val mapper = ObjectMapper()
+  private val instances = mutable.Map[String, InstanceMetadata]()
+  private var redisPrefix = "nowchess"
+
+  def setRedisPrefix(prefix: String): Unit =
+    redisPrefix = prefix
+
+  def getInstance(instanceId: String): Option[InstanceMetadata] =
+    instances.get(instanceId)
+
+  def getAllInstances: List[InstanceMetadata] =
+    instances.values.toList
+
+  def listInstancesFromRedis: List[InstanceMetadata] =
+    val pattern = s"$redisPrefix:instances:*"
+    val keys = redissonClient.getKeys.getKeysByPattern(pattern, 100)
+    keys.asScala.flatMap { key =>
+      val bucket = redissonClient.getBucket[String](key)
+      val value = bucket.getAndDelete()
+      if value != null then
+        try
+          Some(mapper.readValue(value, classOf[InstanceMetadata]))
+        catch
+          case _: Exception => None
+      else
+        None
+    }.toList
+
+  def updateInstanceFromRedis(instanceId: String): Unit =
+    val key = s"$redisPrefix:instances:$instanceId"
+    val bucket = redissonClient.getBucket[String](key)
+    val value = bucket.get()
+    if value != null then
+      try
+        val metadata = mapper.readValue(value, classOf[InstanceMetadata])
+        instances(instanceId) = metadata
+      catch
+        case _: Exception => ()
+
+  def markInstanceDead(instanceId: String): Unit =
+    instances.get(instanceId).foreach { inst =>
+      val dead = inst.copy(state = "DEAD")
+      instances(instanceId) = dead
+    }
+
+  def removeInstance(instanceId: String): Unit =
+    instances.remove(instanceId)
@@ -0,0 +1,44 @@
+package de.nowchess.coordinator.service
+
+import jakarta.enterprise.context.ApplicationScoped
+import jakarta.inject.Inject
+import de.nowchess.coordinator.config.CoordinatorConfig
+import org.jboss.logging.Logger
+import scala.concurrent.duration.*
+
+@ApplicationScoped
+class LoadBalancer:
+  @Inject
+  private var config: CoordinatorConfig = _
+
+  @Inject
+  private var instanceRegistry: InstanceRegistry = _
+
+  private val log = Logger.getLogger(classOf[LoadBalancer])
+  private var lastRebalanceTime = 0L
+
+  def shouldRebalance: Boolean =
+    val now = System.currentTimeMillis()
+    val minInterval = config.rebalanceMinInterval.toMillis
+    if now - lastRebalanceTime < minInterval then
+      return false
+
+    val instances = instanceRegistry.getAllInstances
+    if instances.isEmpty then return false
+
+    val loads = instances.map(_.subscriptionCount)
+    val maxLoad = loads.max
+    val minLoad = loads.min
+    val avgLoad = loads.sum.toDouble / loads.size
+
+    val exceededMax = maxLoad > config.maxGamesPerCore
+    val deviationPercent = 100.0 * (maxLoad - avgLoad) / avgLoad
+    val exceededDeviation = maxLoad > avgLoad && deviationPercent > config.maxDeviationPercent && (maxLoad - minLoad) > 50
+
+    exceededMax || exceededDeviation
+
+  def rebalance: Unit =
+    log.info("Starting rebalance")
+    lastRebalanceTime = System.currentTimeMillis()
+    // Placeholder: actual rebalance logic will be implemented
+    log.info("Rebalance completed")
@@ -0,0 +1,57 @@
+syntax = "proto3";
+
+package de.nowchess.coordinator;
+
+service CoordinatorService {
+  rpc HeartbeatStream(stream HeartbeatFrame) returns (stream CoordinatorCommand);
+  rpc BatchResubscribeGames(BatchResubscribeRequest) returns (BatchResubscribeResponse);
+  rpc UnsubscribeGames(UnsubscribeGamesRequest) returns (UnsubscribeGamesResponse);
+  rpc EvictGames(EvictGamesRequest) returns (EvictGamesResponse);
+  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 {
+  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 = 1;
+}
@@ -11,6 +11,9 @@ quarkus:
      io-grpc:
        host: localhost
        port: 8081
+      coordinator-grpc:
+        host: localhost
+        port: 9086
    server:
      use-separate-server: false

@@ -20,6 +23,13 @@ nowchess:
    port: 6379
    prefix: nowchess

+  coordinator:
+    host: localhost
+    grpc-port: 9086
+    stream-heartbeat-interval: 200ms
+    redis-heartbeat-interval: 2s
+    instance-id: ${HOSTNAME:local}-${quarkus.uuid}
+
 "%dev":
  mp:
    jwt:
@@ -72,6 +82,9 @@ nowchess:
        io-grpc:
          host: ${IO_SERVICE_HOST}
          port: ${IO_SERVICE_GRPC_PORT:9081}
+        coordinator-grpc:
+          host: ${COORDINATOR_SERVICE_HOST:localhost}
+          port: ${COORDINATOR_SERVICE_GRPC_PORT:9086}
    rest-client:
      io-service:
        url: ${IO_SERVICE_URL}
@@ -84,3 +97,10 @@ nowchess:
      host: ${REDIS_HOST}
      port: ${REDIS_PORT:6379}
      prefix: ${REDIS_PREFIX:nowchess}
+
+    coordinator:
+      host: ${COORDINATOR_SERVICE_HOST:localhost}
+      grpc-port: ${COORDINATOR_SERVICE_GRPC_PORT:9086}
+      stream-heartbeat-interval: 200ms
+      redis-heartbeat-interval: 2s
+      instance-id: ${HOSTNAME:local}-${quarkus.uuid}
@@ -0,0 +1,67 @@
+package de.nowchess.chess.grpc
+
+import jakarta.enterprise.context.ApplicationScoped
+import jakarta.inject.Inject
+import de.nowchess.coordinator.CoordinatorServiceGrpc
+import de.nowchess.coordinator.{
+  BatchResubscribeRequest,
+  BatchResubscribeResponse,
+  UnsubscribeGamesRequest,
+  UnsubscribeGamesResponse,
+  EvictGamesRequest,
+  EvictGamesResponse,
+  DrainInstanceRequest,
+  DrainInstanceResponse
+}
+import de.nowchess.chess.redis.GameRedisSubscriberManager
+import io.grpc.stub.StreamObserver
+import scala.jdk.CollectionConverters.*
+
+@ApplicationScoped
+class CoordinatorServiceHandler extends CoordinatorServiceGrpc.CoordinatorServiceImplBase:
+  @Inject
+  private var gameSubscriberManager: GameRedisSubscriberManager = _
+
+  override def batchResubscribeGames(
+    request: BatchResubscribeRequest,
+    responseObserver: StreamObserver[BatchResubscribeResponse]
+  ): Unit =
+    val count = gameSubscriberManager.batchResubscribeGames(request.getGameIdsList)
+    val response = BatchResubscribeResponse.newBuilder()
+      .setSubscribedCount(count)
+      .build()
+    responseObserver.onNext(response)
+    responseObserver.onCompleted()
+
+  override def unsubscribeGames(
+    request: UnsubscribeGamesRequest,
+    responseObserver: StreamObserver[UnsubscribeGamesResponse]
+  ): Unit =
+    val count = gameSubscriberManager.unsubscribeGames(request.getGameIdsList)
+    val response = UnsubscribeGamesResponse.newBuilder()
+      .setUnsubscribedCount(count)
+      .build()
+    responseObserver.onNext(response)
+    responseObserver.onCompleted()
+
+  override def evictGames(
+    request: EvictGamesRequest,
+    responseObserver: StreamObserver[EvictGamesResponse]
+  ): Unit =
+    val count = gameSubscriberManager.evictGames(request.getGameIdsList)
+    val response = EvictGamesResponse.newBuilder()
+      .setEvictedCount(count)
+      .build()
+    responseObserver.onNext(response)
+    responseObserver.onCompleted()
+
+  override def drainInstance(
+    request: DrainInstanceRequest,
+    responseObserver: StreamObserver[DrainInstanceResponse]
+  ): Unit =
+    gameSubscriberManager.drainInstance()
+    val response = DrainInstanceResponse.newBuilder()
+      .setGamesMigrated(0)
+      .build()
+    responseObserver.onNext(response)
+    responseObserver.onCompleted()
@@ -7,6 +7,7 @@ import de.nowchess.chess.grpc.IoGrpcClientWrapper
 import de.nowchess.chess.observer.Observer
 import de.nowchess.chess.registry.GameRegistry
 import de.nowchess.chess.resource.GameDtoMapper
+import de.nowchess.chess.service.InstanceHeartbeatService
 import jakarta.annotation.PreDestroy
 import jakarta.enterprise.context.ApplicationScoped
 import jakarta.inject.Inject
@@ -25,6 +26,7 @@ class GameRedisSubscriberManager:
  @Inject var objectMapper: ObjectMapper = uninitialized
  @Inject var redisConfig: RedisConfig = uninitialized
  @Inject var ioClient: IoGrpcClientWrapper = uninitialized
+  @Inject(optional = true) var heartbeatService: InstanceHeartbeatService = uninitialized
  // scalafix:on DisableSyntax.var

  private val c2sListeners = new ConcurrentHashMap[String, Int]()
@@ -50,6 +52,9 @@ class GameRedisSubscriberManager:
      val obs = new GameRedisPublisher(gameId, registry, redisson, objectMapper, s2cTopicName(gameId), writebackFn, ioClient, unsubscribeGame)
      s2cObservers.put(gameId, obs)
      registry.get(gameId).foreach(_.engine.subscribe(obs))
+
+      if heartbeatService != null then
+        heartbeatService.addGameSubscription(gameId)
    catch
      case e: Exception =>
        System.err.println(s"Warning: Redis subscription failed for game $gameId: ${e.getMessage}")
@@ -63,6 +68,9 @@ class GameRedisSubscriberManager:
      registry.get(gameId).foreach(_.engine.unsubscribe(obs))
    }

+    if heartbeatService != null then
+      heartbeatService.removeGameSubscription(gameId)
+
  private def handleC2sMessage(gameId: String, msg: String): Unit =
    parseC2sMessage(msg) match
      case Some(C2sMessage.Connected)   => handleConnected(gameId)
@@ -92,6 +100,35 @@ class GameRedisSubscriberManager:
      }
    }

+  def batchResubscribeGames(gameIds: java.util.List[String]): Int =
+    var count = 0
+    gameIds.forEach { gameId =>
+      subscribeGame(gameId)
+      count += 1
+    }
+    count
+
+  def unsubscribeGames(gameIds: java.util.List[String]): Int =
+    var count = 0
+    gameIds.forEach { gameId =>
+      unsubscribeGame(gameId)
+      count += 1
+    }
+    count
+
+  def evictGames(gameIds: java.util.List[String]): Int =
+    var count = 0
+    gameIds.forEach { gameId =>
+      unsubscribeGame(gameId)
+      registry.remove(gameId)
+      count += 1
+    }
+    count
+
+  def drainInstance(): Unit =
+    val gameIds = new java.util.ArrayList(c2sListeners.keySet())
+    gameIds.forEach(unsubscribeGame)
+
  @PreDestroy
  def cleanup(): Unit =
    c2sListeners.forEach((gameId, listenerId) =>
@@ -0,0 +1,186 @@
+package de.nowchess.chess.service
+
+import jakarta.enterprise.context.ApplicationScoped
+import jakarta.enterprise.event.Observes
+import jakarta.inject.Inject
+import io.quarkus.runtime.StartupEvent
+import io.quarkus.runtime.ShutdownEvent
+import io.quarkus.grpc.GrpcClient
+import org.redisson.api.RedissonClient
+import scala.concurrent.duration.*
+import java.util.concurrent.{Executors, TimeUnit}
+import java.net.InetAddress
+import com.fasterxml.jackson.databind.ObjectMapper
+import org.jboss.logging.Logger
+import de.nowchess.coordinator.{HeartbeatFrame, CoordinatorServiceGrpc}
+import de.nowchess.coordinator.CoordinatorServiceGrpc.CoordinatorServiceStub
+import io.grpc.stub.StreamObserver
+import scala.jdk.FutureConverters.*
+
+@ApplicationScoped
+class InstanceHeartbeatService:
+  @Inject
+  private var redissonClient: RedissonClient = _
+
+  @GrpcClient("coordinator-grpc")
+  private var coordinatorStub: CoordinatorServiceStub = _
+
+  private val log = Logger.getLogger(classOf[InstanceHeartbeatService])
+  private val mapper = ObjectMapper()
+
+  private var instanceId = ""
+  private var redisPrefix = "nowchess"
+  private var streamObserver: Option[StreamObserver[HeartbeatFrame]] = None
+  private var heartbeatExecutor = Executors.newScheduledThreadPool(1)
+  private var redisHeartbeatExecutor = Executors.newScheduledThreadPool(1)
+  private var subscriptionCount = 0
+  private var localCacheSize = 0
+
+  def onStart(@Observes event: StartupEvent): Unit =
+    try
+      generateInstanceId()
+      initializeHeartbeatStream()
+      scheduleHeartbeats()
+      log.infof("Instance heartbeat service started with ID: %s", instanceId)
+    catch
+      case ex: Exception =>
+        log.errorf(ex, "Failed to start instance heartbeat service")
+
+  def onShutdown(@Observes event: ShutdownEvent): Unit =
+    try
+      cleanup()
+      log.info("Instance heartbeat service stopped")
+    catch
+      case ex: Exception =>
+        log.errorf(ex, "Error during heartbeat service shutdown")
+
+  def setRedisPrefix(prefix: String): Unit =
+    redisPrefix = prefix
+
+  def setSubscriptionCount(count: Int): Unit =
+    subscriptionCount = count
+
+  def setLocalCacheSize(count: Int): Unit =
+    localCacheSize = count
+
+  def addGameSubscription(gameId: String): Unit =
+    val setKey = s"$redisPrefix:instance:$instanceId:games"
+    val gameSet = redissonClient.getSet[String](setKey)
+    gameSet.add(gameId)
+    subscriptionCount += 1
+
+  def removeGameSubscription(gameId: String): Unit =
+    val setKey = s"$redisPrefix:instance:$instanceId:games"
+    val gameSet = redissonClient.getSet[String](setKey)
+    gameSet.remove(gameId)
+    subscriptionCount = Math.max(0, subscriptionCount - 1)
+
+  private def generateInstanceId(): Unit =
+    val hostname = try
+      InetAddress.getLocalHost.getHostName
+    catch
+      case _: Exception => "unknown"
+
+    val uuid = java.util.UUID.randomUUID().toString.take(8)
+    instanceId = s"$hostname-$uuid"
+
+  private def initializeHeartbeatStream(): Unit =
+    try
+      val responseObserver = new StreamObserver[de.nowchess.coordinator.CoordinatorCommand]:
+        override def onNext(value: de.nowchess.coordinator.CoordinatorCommand): Unit =
+          log.debugf("Received coordinator command: %s", value.getType)
+
+        override def onError(t: Throwable): Unit =
+          log.warnf(t, "Heartbeat stream error")
+          // Reconnect on error
+          () // Placeholder for reconnect logic
+
+        override def onCompleted: Unit =
+          log.info("Heartbeat stream completed")
+
+      streamObserver = Some(coordinatorStub.heartbeatStream(responseObserver))
+      log.info("Connected to coordinator heartbeat stream")
+    catch
+      case ex: Exception =>
+        log.warnf(ex, "Failed to connect to coordinator")
+        streamObserver = None
+
+  private def scheduleHeartbeats(): Unit =
+    // Send heartbeat every 200ms
+    heartbeatExecutor.scheduleAtFixedRate(
+      () => sendHeartbeat(),
+      0,
+      200,
+      TimeUnit.MILLISECONDS
+    )
+
+    // Refresh Redis TTL every 2s
+    redisHeartbeatExecutor.scheduleAtFixedRate(
+      () => refreshRedisHeartbeat(),
+      0,
+      2,
+      TimeUnit.SECONDS
+    )
+
+  private def sendHeartbeat(): Unit =
+    streamObserver.foreach { observer =>
+      try
+        val frame = HeartbeatFrame.newBuilder()
+          .setInstanceId(instanceId)
+          .setHostname(getHostname)
+          .setHttpPort(8080) // Placeholder, should be configurable
+          .setGrpcPort(9080) // Placeholder
+          .setSubscriptionCount(subscriptionCount)
+          .setLocalCacheSize(localCacheSize)
+          .setTimestampMillis(System.currentTimeMillis())
+          .build()
+        observer.onNext(frame)
+      catch
+        case ex: Exception =>
+          log.warnf(ex, "Failed to send heartbeat frame")
+    }
+
+  private def refreshRedisHeartbeat(): Unit =
+    try
+      val key = s"$redisPrefix:instances:$instanceId"
+      val bucket = redissonClient.getBucket[String](key)
+
+      val metadata = Map(
+        "instanceId" -> instanceId,
+        "hostname" -> getHostname,
+        "httpPort" -> 8080,
+        "grpcPort" -> 9080,
+        "subscriptionCount" -> subscriptionCount,
+        "localCacheSize" -> localCacheSize,
+        "lastHeartbeat" -> java.time.Instant.now().toString,
+        "state" -> "HEALTHY"
+      )
+
+      val json = mapper.writeValueAsString(metadata)
+      bucket.set(json, 5, TimeUnit.SECONDS) // 5-second TTL, refreshed every 2s
+    catch
+      case ex: Exception =>
+        log.warnf(ex, "Failed to refresh Redis heartbeat")
+
+  private def getHostname: String =
+    try
+      InetAddress.getLocalHost.getHostName
+    catch
+      case _: Exception => "unknown"
+
+  private def cleanup(): Unit =
+    streamObserver.foreach(_.onCompleted())
+    streamObserver = None
+
+    val key = s"$redisPrefix:instances:$instanceId"
+    redissonClient.getBucket[String](key).delete()
+
+    val setKey = s"$redisPrefix:instance:$instanceId:games"
+    redissonClient.getSet[String](setKey).delete()
+
+    heartbeatExecutor.shutdown()
+    redisHeartbeatExecutor.shutdown()
+    if !heartbeatExecutor.awaitTermination(5, TimeUnit.SECONDS) then
+      heartbeatExecutor.shutdownNow()
+    if !redisHeartbeatExecutor.awaitTermination(5, TimeUnit.SECONDS) then
+      redisHeartbeatExecutor.shutdownNow()