feat: NCS-82 add Swiss-system tournament module #55

Merged
lq64 merged 11 commits from feat/NCS-82 into main 2026-06-09 15:09:53 +02:00
5 changed files with 280 additions and 3 deletions
Showing only changes of commit 311da8efba - Show all commits
2
@@ -217,6 +217,8 @@ class AccountService:
bot.name = name
bot.createdAt = Instant.now()
officialBotAccountRepository.persist(bot)
bot.token = generateBotToken(bot.id, bot.name)
officialBotAccountRepository.persist(bot)
log.infof("Auto-registered official bot: %s", name)
}
1
@@ -22,7 +22,7 @@ import org.jboss.logging.Logger
import scala.compiletime.uninitialized
import scala.jdk.CollectionConverters.*
import scala.util.Try
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.{ConcurrentHashMap, ExecutorService, Executors}
import java.util.function.Consumer
@ApplicationScoped
@@ -46,6 +46,10 @@ class GameRedisSubscriberManager:
private val c2sListeners = new ConcurrentHashMap[String, ReactivePubSubCommands.ReactiveRedisSubscriber]()
private val s2cObservers = new ConcurrentHashMap[String, Observer]()
// Per-game single-thread executor so c2s messages are handled off the Vert.x
// event loop (handleConnected/handleMove make blocking gRPC + Redis calls) while
// staying ordered per game.
private val c2sExecutors = new ConcurrentHashMap[String, ExecutorService]()
// scalafix:off DisableSyntax.var
private var clockExpireSubscriber: Option[ReactivePubSubCommands.ReactiveRedisSubscriber] = None
@@ -95,7 +99,14 @@ class GameRedisSubscriberManager:
obs.emitInitialWriteback()
heartbeatServiceOpt.foreach(_.addGameSubscription(gameId))
val handler: Consumer[String] = msg => handleC2sMessage(gameId, msg)
val executor = c2sExecutors.computeIfAbsent(gameId, _ => Executors.newSingleThreadExecutor())
Outdated
Review

newSingleThreadExecutor() uses an unbounded queue. Under a slow game handler, tasks will queue indefinitely. Use new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(1000), r -> ..., new DiscardOldestPolicy()) to bound memory.

newSingleThreadExecutor() uses an unbounded queue. Under a slow game handler, tasks will queue indefinitely. Use new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(1000), r -> ..., new DiscardOldestPolicy()) to bound memory.
val handler: Consumer[String] = msg =>
val task = new Runnable:
def run(): Unit =
try handleC2sMessage(gameId, msg)
catch case ex: Exception => log.warnf(ex, "Error handling c2s message for game %s", gameId)
Try(executor.execute(task))
Outdated
Review

Swallowing a rejected RejectedExecutionException silently drops moves. At minimum log the rejection so it shows up in monitoring.

Swallowing a rejected RejectedExecutionException silently drops moves. At minimum log the rejection so it shows up in monitoring.
()
try
val subscriber = reactiveRedis
.pubsub(classOf[String])
1
@@ -113,6 +124,7 @@ class GameRedisSubscriberManager:
Option(s2cObservers.remove(gameId)).foreach { obs =>
registry.get(gameId).foreach(_.engine.unsubscribe(obs))
}
Option(c2sExecutors.remove(gameId)).foreach(_.shutdownNow())
heartbeatServiceOpt.foreach(_.removeGameSubscription(gameId))
log.debugf("Unsubscribed from game %s", gameId)
1
@@ -187,3 +199,4 @@ class GameRedisSubscriberManager:
clockExpireSubscriber.foreach(_.unsubscribe(clockExpireChannel).await().indefinitely())
c2sListeners.forEach((gameId, subscriber) => subscriber.unsubscribe(c2sTopic(gameId)).await().indefinitely())
s2cObservers.forEach((gameId, obs) => registry.get(gameId).foreach(_.engine.unsubscribe(obs)))
c2sExecutors.forEach((_, executor) => executor.shutdownNow())
@@ -0,0 +1,35 @@
package de.nowchess.bot.service
import com.fasterxml.jackson.databind.ObjectMapper
import scala.util.Try
final case class TournamentBotConfig(
serverUrl: String,
tournamentId: String,
token: String,
botId: String,
difficulty: String,
)
object TournamentBotConfig:
private val mapper = new ObjectMapper()
def fromEnv(env: Map[String, String]): Option[TournamentBotConfig] =
for
tournamentId <- env.get("TOURNAMENT_ID").filter(_.nonEmpty)
token <- env.get("TOURNAMENT_BOT_TOKEN").filter(_.nonEmpty)
botId <- jwtSubject(token)
serverUrl = env.getOrElse("TOURNAMENT_SERVER_URL", "http://localhost:8089")
difficulty = env.getOrElse("TOURNAMENT_BOT_DIFFICULTY", "medium")
yield TournamentBotConfig(serverUrl, tournamentId, token, botId, difficulty)
def jwtSubject(token: String): Option[String] =
Try {
val parts = token.split("\\.")
if parts.length >= 2 then
val payload = new String(java.util.Base64.getUrlDecoder.decode(parts(1)))
val sub = mapper.readTree(payload).path("sub").asText()
Option(sub).filter(_.nonEmpty)
else None
}.toOption.flatten
@@ -0,0 +1,219 @@
package de.nowchess.bot.service
Outdated
Review

The entire file operates on a daemon thread with Thread.sleep(5000) reconnect. This bypasses Quarkus CDI lifecycle: it won't receive the shutdown signal, won't drain in-flight work on @PreDestroy, and the flat 5-second backoff will hammer the server if the tournament endpoint is persistently unavailable. Use @Scheduled or a Vert.x timer with exponential backoff, and track the thread reference so it can be interrupted on graceful shutdown.

The activeGames set that prevents duplicate game processing is never trimmed when a game ends. In a long-lived process across many rounds the set grows without bound. Remove game IDs from activeGames when a terminal game state is received.

The entire file operates on a daemon thread with Thread.sleep(5000) reconnect. This bypasses Quarkus CDI lifecycle: it won't receive the shutdown signal, won't drain in-flight work on @PreDestroy, and the flat 5-second backoff will hammer the server if the tournament endpoint is persistently unavailable. Use @Scheduled or a Vert.x timer with exponential backoff, and track the thread reference so it can be interrupted on graceful shutdown. The activeGames set that prevents duplicate game processing is never trimmed when a game ends. In a long-lived process across many rounds the set grows without bound. Remove game IDs from activeGames when a terminal game state is received.
import com.fasterxml.jackson.databind.{JsonNode, ObjectMapper}
import de.nowchess.api.move.{Move, MoveType, PromotionPiece}
import de.nowchess.bot.{Bot, BotController}
import de.nowchess.io.fen.FenParser
import io.quarkus.runtime.Startup
import jakarta.annotation.{PostConstruct, PreDestroy}
import jakarta.enterprise.context.ApplicationScoped
import jakarta.inject.Inject
import jakarta.ws.rs.client.{Client, ClientBuilder, Entity}
import jakarta.ws.rs.core.MediaType
import org.jboss.logging.Logger
import scala.compiletime.uninitialized
import scala.jdk.CollectionConverters.*
import scala.util.{Failure, Success, Try}
import java.io.{BufferedReader, InputStream, InputStreamReader}
import java.util.concurrent.{ConcurrentHashMap, ExecutorService, Executors}
@Startup
@ApplicationScoped
class TournamentBotGamePlayer:
private val log = Logger.getLogger(classOf[TournamentBotGamePlayer])
// scalafix:off DisableSyntax.var
@Inject var objectMapper: ObjectMapper = uninitialized
@Inject var botController: BotController = uninitialized
// scalafix:on DisableSyntax.var
private val client: Client = ClientBuilder.newClient()
private val workers: ExecutorService = Executors.newCachedThreadPool()
private val activeGames = ConcurrentHashMap.newKeySet[String]()
private val config = TournamentBotConfig.fromEnv(System.getenv().asScala.toMap)
// scalafix:off DisableSyntax.var
@volatile private var running = true
// scalafix:on DisableSyntax.var
@PostConstruct
def initialize(): Unit =
config match
case None =>
log.info("Tournament bot disabled — set TOURNAMENT_ID and TOURNAMENT_BOT_TOKEN to enable")
case Some(cfg) =>
log.infof("Tournament bot enabled — server=%s tournament=%s bot=%s", cfg.serverUrl, cfg.tournamentId, cfg.botId)
val thread = new Thread(() => connect(cfg), s"TournamentBot-${cfg.tournamentId}")
thread.setDaemon(true)
thread.start()
@PreDestroy
def cleanup(): Unit =
running = false
workers.shutdownNow()
Try(client.close())
log.info("Tournament bot stopped")
private def connect(cfg: TournamentBotConfig): Unit =
if join(cfg) then
while running do
Try(streamEvents(cfg)) match
case Failure(ex) => log.warnf(ex, "Tournament event stream dropped — reconnecting"); sleep(5000)
case Success(_) => sleep(2000)
private def join(cfg: TournamentBotConfig): Boolean =
Try {
val response = authed(cfg, target(cfg).path("join"))
.post(Entity.entity("", MediaType.APPLICATION_JSON))
val ok = response.getStatus == 200
if ok then log.infof("Joined tournament %s", cfg.tournamentId)
else log.errorf("Failed to join tournament %s — status %d", cfg.tournamentId, response.getStatus)
response.close()
ok
}.getOrElse { log.error("Join request failed"); false }
private def streamEvents(cfg: TournamentBotConfig): Unit =
val response = authed(cfg, target(cfg).path("stream"))
.header("Accept", "application/x-ndjson")
.get()
if response.getStatus != 200 then
log.warnf("Tournament stream returned status %d", response.getStatus)
response.close()
sleep(5000)
else
log.infof("Listening to tournament %s event stream", cfg.tournamentId)
forEachLine(response.readEntity(classOf[InputStream])): line =>
parse(line).foreach: node =>
if node.path("type").asText() == "gameStart" then onGameStart(cfg, node.path("gameId").asText())
private def onGameStart(cfg: TournamentBotConfig, gameId: String): Unit =
if gameId.nonEmpty && activeGames.add(gameId) then
workers.submit(new Runnable { def run(): Unit = playGame(cfg, gameId) })
()
private def playGame(cfg: TournamentBotConfig, gameId: String): Unit =
Try {
colorFor(cfg, gameId) match
case None =>
log.debugf("Game %s is not ours — ignoring", gameId)
activeGames.remove(gameId)
case Some(color) =>
log.infof("Playing game %s as %s", gameId, color)
val stream = openGameStream(cfg, gameId)
maybeMoveFromCurrentState(cfg, gameId, color)
if stream != null then consumeGameStream(cfg, gameId, color, stream)
activeGames.remove(gameId)
} match
case Failure(ex) => log.errorf(ex, "Game %s crashed", gameId); activeGames.remove(gameId)
case Success(_) => ()
private def colorFor(cfg: TournamentBotConfig, gameId: String): Option[String] =
fetchGame(cfg, gameId).flatMap: game =>
val white = game.path("white").path("id").asText()
val black = game.path("black").path("id").asText()
if white == cfg.botId then Some("white")
else if black == cfg.botId then Some("black")
else None
private def maybeMoveFromCurrentState(cfg: TournamentBotConfig, gameId: String, color: String): Unit =
fetchGame(cfg, gameId).foreach: game =>
maybeMove(cfg, gameId, color, game.path("turn").asText(), game.path("status").asText(), game.path("fen").asText())
private def consumeGameStream(cfg: TournamentBotConfig, gameId: String, color: String, stream: InputStream): Unit =
val reader = new BufferedReader(new InputStreamReader(stream))
// scalafix:off DisableSyntax.var
var done = false
var line = reader.readLine()
// scalafix:on DisableSyntax.var
while line != null && running && !done do
parse(line).foreach: node =>
node.path("type").asText() match
case "move" => maybeMove(cfg, gameId, color, node.path("turn").asText(), "ongoing", node.path("fen").asText())
case "gameEnd" => log.infof("Game %s ended — status=%s", gameId, node.path("status").asText()); done = true
case _ => ()
line = reader.readLine()
private def maybeMove(
cfg: TournamentBotConfig,
gameId: String,
color: String,
turn: String,
status: String,
fen: String,
): Unit =
if turn == color && status == "ongoing" && fen.nonEmpty then
computeUci(cfg, fen) match
case None => log.warnf("No move found for game %s (fen=%s)", gameId, fen)
case Some(uci) => submitMove(cfg, gameId, uci)
private def computeUci(cfg: TournamentBotConfig, fen: String): Option[String] =
FenParser.parseFen(fen) match
case Left(err) => log.warnf("FEN parse failed: %s (%s)", fen, err.toString); None
case Right(context) => engine(cfg).apply(context).map(toUci)
private def submitMove(cfg: TournamentBotConfig, gameId: String, uci: String): Unit =
Try {
val response = authed(cfg, target(cfg).path("game").path(gameId).path("move").path(uci))
.post(Entity.entity("", MediaType.APPLICATION_JSON))
if response.getStatus == 200 then log.infof("Played %s in game %s", uci, gameId)
else log.warnf("Move %s rejected in game %s — status %d", uci, gameId, response.getStatus)
response.close()
} match
case Failure(ex) => log.errorf(ex, "Error submitting move %s in game %s", uci, gameId)
case Success(_) => ()
private def fetchGame(cfg: TournamentBotConfig, gameId: String): Option[JsonNode] =
Try {
val response = target(cfg).path("game").path(gameId).request(MediaType.APPLICATION_JSON).get()
val node = if response.getStatus == 200 then Some(response.readEntity(classOf[JsonNode])) else None
response.close()
node
}.getOrElse(None)
private def openGameStream(cfg: TournamentBotConfig, gameId: String): InputStream =
Try {
val response = authed(cfg, target(cfg).path("game").path(gameId).path("stream"))
.header("Accept", "application/x-ndjson")
.get()
if response.getStatus == 200 then response.readEntity(classOf[InputStream])
else { log.warnf("Game stream %s returned status %d", gameId, response.getStatus); response.close(); null }
}.getOrElse(null)
private def engine(cfg: TournamentBotConfig): Bot =
botController.getBot(cfg.difficulty).orElse(botController.getBot("medium")).get
private def target(cfg: TournamentBotConfig) =
client.target(cfg.serverUrl).path("api").path("tournament").path(cfg.tournamentId)
private def authed(cfg: TournamentBotConfig, t: jakarta.ws.rs.client.WebTarget) =
t.request(MediaType.APPLICATION_JSON).header("Authorization", s"Bearer ${cfg.token}")
private def parse(line: String): Option[JsonNode] =
val trimmed = line.trim
if trimmed.isEmpty then None else Try(objectMapper.readTree(trimmed)).toOption
private def forEachLine(stream: InputStream)(handle: String => Unit): Unit =
val reader = new BufferedReader(new InputStreamReader(stream))
// scalafix:off DisableSyntax.var
var line: String = reader.readLine()
// scalafix:on DisableSyntax.var
while line != null && running do
Try(handle(line)).failed.foreach(ex => log.warnf(ex, "Error handling stream line"))
line = reader.readLine()
private def toUci(move: Move): String =
val base = s"${move.from}${move.to}"
move.moveType match
case MoveType.Promotion(piece) => base + promotionChar(piece)
case _ => base
private def promotionChar(piece: PromotionPiece): String =
piece match
case PromotionPiece.Knight => "n"
case PromotionPiece.Bishop => "b"
case PromotionPiece.Rook => "r"
case PromotionPiece.Queen => "q"
private def sleep(ms: Long): Unit = Try(Thread.sleep(ms))
@@ -49,7 +49,8 @@ class GameResultStreamListener:
case Success(_) => ()
private def pollLoop(): Unit =
while true do
var running = true
while running do
Try {
val messages = redis.stream(classOf[String]).xreadgroup(
groupName,
@@ -60,9 +61,16 @@ class GameResultStreamListener:
)
if messages != null then messages.forEach(msg => handleMessage(msg))
} match
case Failure(ex) if isInterrupted(ex) =>
Thread.currentThread().interrupt()
running = false
case Failure(ex) => log.warnf(ex, "Error in result poll loop")
case Success(_) => ()
private def isInterrupted(ex: Throwable): Boolean =
ex.isInstanceOf[InterruptedException] ||
(ex.getCause != null && ex.getCause.isInstanceOf[InterruptedException])
private def handleMessage(msg: StreamMessage[String, String, String]): Unit =
val json = msg.payload().get("data")
Try(objectMapper.readValue(json, classOf[GameWritebackEventDto])) match