diff options
11 files changed, 161 insertions, 22 deletions
diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/LunaticChat.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/LunaticChat.kt index 3041759..f39e5d4 100644 --- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/LunaticChat.kt +++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/LunaticChat.kt @@ -68,7 +68,7 @@ class LunaticChat : // Initialize plugin coroutine scope pluginScope = PluginCoroutineScope(logger) - deliveryQueue = PerPlayerWorkQueue(pluginScope.scope) + deliveryQueue = PerPlayerWorkQueue(pluginScope.scope, logger) // Initialize all services serviceInitializer = diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/PerPlayerWorkQueue.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/PerPlayerWorkQueue.kt index 392eea8..0769823 100644 --- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/PerPlayerWorkQueue.kt +++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/PerPlayerWorkQueue.kt @@ -1,11 +1,14 @@ package dev.m1sk9.lunaticChat.paper +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.SendChannel import kotlinx.coroutines.launch import java.util.UUID import java.util.concurrent.ConcurrentHashMap +import java.util.logging.Level +import java.util.logging.Logger /** * Runs work submitted for a player one item at a time, in the order it was submitted. @@ -20,6 +23,7 @@ import java.util.concurrent.ConcurrentHashMap */ class PerPlayerWorkQueue( private val scope: CoroutineScope, + private val logger: Logger, ) { private val queues = ConcurrentHashMap<UUID, SendChannel<suspend () -> Unit>>() @@ -30,7 +34,13 @@ class PerPlayerWorkQueue( playerId: UUID, work: suspend () -> Unit, ) { - queues.computeIfAbsent(playerId) { startWorker() }.trySend(work) + val accepted = queues.computeIfAbsent(playerId) { startWorker() }.trySend(work).isSuccess + if (!accepted) { + // Reachable once the scope is cancelled at shutdown, or if the player's queue is + // released in the same tick as their command. Dropping a message in silence is worse + // than saying so. + logger.warning("Discarded queued work for player $playerId: their queue is closed") + } } /** @@ -46,7 +56,16 @@ class PerPlayerWorkQueue( val channel = Channel<suspend () -> Unit>(Channel.UNLIMITED) scope.launch { for (work in channel) { - work() + // One failed message must not end the loop. If it did, the channel would stay in + // `queues` with nothing reading it, so every later message from this player would + // be buffered and never delivered - with no way to recover short of reconnecting. + try { + work() + } catch (e: CancellationException) { + throw e + } catch (e: Throwable) { + logger.log(Level.SEVERE, "Queued work failed", e) + } } } return channel diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/ServiceInitializer.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/ServiceInitializer.kt index 522a1a6..1029c40 100644 --- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/ServiceInitializer.kt +++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/ServiceInitializer.kt @@ -53,6 +53,11 @@ class ServiceInitializer( ) { private val handshakeCompleted = AtomicBoolean(false) + private companion object { + /** Matches the value documented in config.yml. */ + const val DEFAULT_CACHE_SAVE_INTERVAL_SECONDS = 300L + } + /** * Initializes all services in dependency order. * @@ -419,12 +424,20 @@ class ServiceInitializer( val conversionCache = services.conversionCache if (conversionCache != null) { // The periodic task is the only writer besides shutdown, so a non-positive interval - // would both reject the schedule and leave the cache unsaved until the server stops. + // would both be rejected by runAtFixedRate and leave the cache unsaved until the + // server stopped. Fall back to the documented default rather than to one second, + // which would rewrite the whole cache file every tick anyone chatted. + val configuredInterval = configuration.features.japaneseConversion.cacheSaveIntervalSeconds val intervalSeconds = - configuration.features.japaneseConversion - .cacheSaveIntervalSeconds - .toLong() - .coerceAtLeast(1) + if (configuredInterval > 0) { + configuredInterval.toLong() + } else { + logger.warning( + "features.japaneseConversion.cache.saveIntervalSeconds must be positive; " + + "using $DEFAULT_CACHE_SAVE_INTERVAL_SECONDS seconds instead of $configuredInterval", + ) + DEFAULT_CACHE_SAVE_INTERVAL_SECONDS + } plugin.server.asyncScheduler.runAtFixedRate( plugin, { conversionCache.saveToDisk() }, diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/chat/channel/ChannelStorage.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/chat/channel/ChannelStorage.kt index 832f8f3..2030eb8 100644 --- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/chat/channel/ChannelStorage.kt +++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/chat/channel/ChannelStorage.kt @@ -5,7 +5,9 @@ import dev.m1sk9.lunaticChat.engine.exception.ChannelStorageLoadException import dev.m1sk9.lunaticChat.engine.exception.ChannelStorageSaveException import dev.m1sk9.lunaticChat.paper.DebouncedSaver import kotlinx.serialization.json.Json +import java.nio.file.Files import java.nio.file.Path +import java.nio.file.StandardCopyOption import java.util.logging.Logger import kotlin.io.path.bufferedReader import kotlin.io.path.exists @@ -66,9 +68,15 @@ class ChannelStorage( fun saveToDisk(data: ChannelData) { try { val jsonContent = json.encodeToString(ChannelData.serializer(), data) - channelsFile.writeText(jsonContent).also { - logger.fine("Successfully saved channels from ${channelsFile.fileName}.") - } + + // Written to a sibling and moved into place. Bukkit runs onDisable before cancelling + // scheduler tasks, so the shutdown save and a still-pending debounced save can reach + // this at the same time; two truncating writes to the same path would interleave and + // leave channels.json unparseable. + val temporaryFile = channelsFile.resolveSibling("${channelsFile.fileName}.tmp") + temporaryFile.writeText(jsonContent) + Files.move(temporaryFile, channelsFile, StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE) + logger.fine("Successfully saved channels from ${channelsFile.fileName}.") } catch (e: Exception) { throw ChannelStorageSaveException( "Failed to save channels to ${channelsFile.fileName}: ${e.message}", diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/chat/handler/DirectMessageHandler.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/chat/handler/DirectMessageHandler.kt index b3266fb..7e9cf1c 100644 --- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/chat/handler/DirectMessageHandler.kt +++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/chat/handler/DirectMessageHandler.kt @@ -66,6 +66,22 @@ class DirectMessageHandler( } /** + * Records that [sender] messaged a player on another server, so /reply can find them. + * + * Separate from the delivery itself because delivery is queued: /reply reads the target on the + * command thread, so recording it only once the message has been converted and sent would + * leave a window - as long as the conversion timeout - where /reply says there is nobody to + * reply to. + */ + fun recordRemoteRecipient( + sender: Player, + targetName: String, + targetServerName: String, + ) { + lastRecipient[sender.uniqueId] = ReplyTarget.Remote(targetName, targetServerName) + } + + /** * Gets the target to reply to. * First checks if someone has messaged this player, otherwise falls back * to the last person they messaged. Targets that are no longer reachable @@ -167,7 +183,7 @@ class DirectMessageHandler( ?.playMessageSendNotification() } - lastRecipient[sender.uniqueId] = ReplyTarget.Remote(targetName, targetServerName) + recordRemoteRecipient(sender, targetName, targetServerName) return displayMessage } diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/ReplyCommand.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/ReplyCommand.kt index 8f676b3..df5b2a9 100644 --- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/ReplyCommand.kt +++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/ReplyCommand.kt @@ -70,6 +70,7 @@ class ReplyCommand( val recipient = Bukkit.getPlayer(target.uuid) ?: return fail("directMessage.replyTargetNotFound") + dmHandler.recordMessage(sender, recipient) deliveryQueue.submit(sender.uniqueId) { dmHandler.sendDirectMessage(sender, recipient, message) } CommandResult.Success } @@ -77,6 +78,7 @@ class ReplyCommand( val manager = crossServerDirectMessageManager ?: return fail("directMessage.replyTargetNotFound") + dmHandler.recordRemoteRecipient(sender, target.playerName, target.serverName) deliveryQueue.submit(sender.uniqueId) { manager.sendCrossServerMessage(sender, target.playerName, target.serverName, message) } diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommand.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommand.kt index f19afed..5e6e2c7 100644 --- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommand.kt +++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommand.kt @@ -104,6 +104,9 @@ class TellCommand( ) { return fail("directMessage.yourself") } + // Recorded here rather than inside the queued work: /reply reads the target on this + // thread, so it must be visible as soon as /tell returns. + directMessageHandler.recordRemoteRecipient(sender, name, server) deliveryQueue.submit(sender.uniqueId) { manager.sendCrossServerMessage(sender, name, server, message) } return CommandResult.Success } @@ -116,6 +119,7 @@ class TellCommand( return fail("directMessage.yourself") } + directMessageHandler.recordMessage(sender, recipient) deliveryQueue.submit(sender.uniqueId) { directMessageHandler.sendDirectMessage(sender, recipient, message) } return CommandResult.Success diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/converter/RomanjiConverter.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/converter/RomanjiConverter.kt index a50a4aa..7908824 100644 --- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/converter/RomanjiConverter.kt +++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/converter/RomanjiConverter.kt @@ -5,6 +5,8 @@ import dev.m1sk9.lunaticChat.engine.converter.KanaConverter import kotlinx.coroutines.async import kotlinx.coroutines.awaitAll import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.sync.Semaphore +import kotlinx.coroutines.sync.withPermit import java.util.logging.Logger class RomanjiConverter( @@ -12,7 +14,13 @@ class RomanjiConverter( private val apiClient: GoogleIMEClient, private val logger: Logger, private val debugMode: Boolean = false, + maxConcurrentRequests: Int = 4, ) { + // A long message would otherwise open one request per word at once. Google IME answering with + // a rate limit lands in convertWord's catch and degrades silently to hiragana, so it is better + // not to ask that hard in the first place. + private val limiter = Semaphore(maxConcurrentRequests) + /** * Converts the given romaji input to Japanese using the API client. * Utilizes a word-level cache to store and retrieve previous conversion results. @@ -38,12 +46,19 @@ class RomanjiConverter( // The words are independent, and callers convert under a timeout covering the whole // message. Awaiting them one at a time makes an N-word message cost N round trips, so a // long message runs out of budget after the first word or two. - val results = + // + // Converted once per distinct word: the sequential version got that for free because it + // cached each word before looking up the next, and a line that repeats a word should not + // ask the API twice for it. + val converted = coroutineScope { - words.map { word -> async { convertWord(word) } }.awaitAll() - } + words + .distinct() + .map { word -> async { word to limiter.withPermit { convertWord(word) } } } + .awaitAll() + }.toMap() - return results.joinToString(" ") + return words.joinToString(" ") { converted.getValue(it) } } private suspend fun convertWord(word: String): String { diff --git a/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/PerPlayerWorkQueueTest.kt b/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/PerPlayerWorkQueueTest.kt index 17341ca..95b7700 100644 --- a/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/PerPlayerWorkQueueTest.kt +++ b/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/PerPlayerWorkQueueTest.kt @@ -1,5 +1,6 @@ package dev.m1sk9.lunaticChat.paper +import dev.m1sk9.lunaticChat.paper.TestUtils import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.delay @@ -27,7 +28,7 @@ class PerPlayerWorkQueueTest { @Test fun `a player's work runs in submission order even when later work is faster`() = runBlocking { - val queue = PerPlayerWorkQueue(CoroutineScope(Dispatchers.Default)) + val queue = PerPlayerWorkQueue(CoroutineScope(Dispatchers.Default), TestUtils.TestLogger()) val completed = ConcurrentLinkedQueue<String>() // The regression this guards: a cached conversion overtaking an uncached one sent @@ -45,7 +46,7 @@ class PerPlayerWorkQueueTest { @Test fun `one player's slow work does not hold up another player`() = runBlocking { - val queue = PerPlayerWorkQueue(CoroutineScope(Dispatchers.Default)) + val queue = PerPlayerWorkQueue(CoroutineScope(Dispatchers.Default), TestUtils.TestLogger()) val completed = ConcurrentLinkedQueue<String>() queue.submit(alice) { @@ -60,7 +61,7 @@ class PerPlayerWorkQueueTest { @Test fun `submit does not block the caller`() { - val queue = PerPlayerWorkQueue(CoroutineScope(Dispatchers.Default)) + val queue = PerPlayerWorkQueue(CoroutineScope(Dispatchers.Default), TestUtils.TestLogger()) val started = ConcurrentLinkedQueue<String>() queue.submit(alice) { @@ -75,7 +76,7 @@ class PerPlayerWorkQueueTest { @Test fun `work already queued still runs after the player is released`() = runBlocking { - val queue = PerPlayerWorkQueue(CoroutineScope(Dispatchers.Default)) + val queue = PerPlayerWorkQueue(CoroutineScope(Dispatchers.Default), TestUtils.TestLogger()) val completed = ConcurrentLinkedQueue<String>() queue.submit(alice) { @@ -91,7 +92,7 @@ class PerPlayerWorkQueueTest { @Test fun `a released player gets a fresh queue if they come back`() = runBlocking { - val queue = PerPlayerWorkQueue(CoroutineScope(Dispatchers.Default)) + val queue = PerPlayerWorkQueue(CoroutineScope(Dispatchers.Default), TestUtils.TestLogger()) val completed = ConcurrentLinkedQueue<String>() queue.submit(alice) { completed.add("before") } @@ -103,4 +104,34 @@ class PerPlayerWorkQueueTest { awaitSize(completed, 2) assertEquals(listOf("before", "after"), completed.toList()) } + + @Test + fun `a failed item does not stop the player's later work`() = + runBlocking { + val queue = PerPlayerWorkQueue(CoroutineScope(Dispatchers.Default), TestUtils.TestLogger()) + val completed = ConcurrentLinkedQueue<String>() + + // Without a guard around each item, the throw would end the consumer loop while its + // channel stayed registered - so everything queued afterwards would be buffered and + // never delivered, and the player would have no way out short of reconnecting. + queue.submit(alice) { error("delivery blew up") } + queue.submit(alice) { completed.add("after-failure") } + + awaitSize(completed, 1) + assertEquals(listOf("after-failure"), completed.toList()) + } + + @Test + fun `a failed item is reported rather than swallowed`() = + runBlocking { + val logger = TestUtils.TestLogger() + val queue = PerPlayerWorkQueue(CoroutineScope(Dispatchers.Default), logger) + val completed = ConcurrentLinkedQueue<String>() + + queue.submit(alice) { error("delivery blew up") } + queue.submit(alice) { completed.add("done") } + awaitSize(completed, 1) + + assertTrue(logger.severeMessages.any { it.contains("Queued work failed") }) + } } diff --git a/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/TestUtils.kt b/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/TestUtils.kt index deabadf..2d2a7c7 100644 --- a/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/TestUtils.kt +++ b/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/TestUtils.kt @@ -16,6 +16,7 @@ import io.mockk.mockk import org.bukkit.entity.Player import org.bukkit.plugin.java.JavaPlugin import java.util.UUID +import java.util.logging.Level import java.util.logging.Logger /** @@ -43,6 +44,20 @@ object TestUtils { severeMessages.add(msg) } + // Logger.severe/warning/info do not route through each other, so code logging with an + // attached throwable would otherwise be invisible to every assertion here. + override fun log( + level: Level, + msg: String, + thrown: Throwable, + ) { + when (level) { + Level.SEVERE -> severeMessages.add(msg) + Level.WARNING -> warningMessages.add(msg) + else -> infoMessages.add(msg) + } + } + fun clear() { infoMessages.clear() warningMessages.clear() diff --git a/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommandTest.kt b/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommandTest.kt index 36b6ff7..d821cf8 100644 --- a/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommandTest.kt +++ b/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommandTest.kt @@ -11,6 +11,7 @@ import dev.m1sk9.lunaticChat.paper.velocity.CrossServerDirectMessageManager import io.mockk.coVerify import io.mockk.every import io.mockk.mockk +import io.mockk.verify import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.test.StandardTestDispatcher @@ -43,7 +44,7 @@ class TellCommandTest { crossServerManager, null, localServerName, - PerPlayerWorkQueue(scope), + PerPlayerWorkQueue(scope, TestUtils.TestLogger()), ) return TellDeps(command, ctx, dmHandler, crossServerManager, sender) } @@ -136,4 +137,19 @@ class TellCommandTest { dispatcher.scheduler.advanceUntilIdle() coVerify { manager.sendCrossServerMessage(deps.sender, "Bob", "survival", "hello") } } + + @Test + fun `the reply target is visible before delivery has run`() { + val dispatcher = StandardTestDispatcher() + val manager = mockk<CrossServerDirectMessageManager>(relaxed = true) + val deps = createCommand(crossServerManager = manager, scope = CoroutineScope(dispatcher)) + + deps.command.execute(deps.ctx, "Bob@survival", "hello") + + // /reply reads the target on the command thread. Recording it inside the queued delivery + // would leave it unset for as long as the conversion takes, so /r straight after /tell + // would report that there is nobody to reply to. + verify { deps.dmHandler.recordRemoteRecipient(deps.sender, "Bob", "survival") } + coVerify(exactly = 0) { manager.sendCrossServerMessage(any(), any(), any(), any()) } + } } |