diff options
10 files changed, 154 insertions, 131 deletions
diff --git a/engine/src/main/kotlin/dev/m1sk9/lunaticChat/engine/protocol/PluginMessageChannel.kt b/engine/src/main/kotlin/dev/m1sk9/lunaticChat/engine/protocol/PluginMessageChannel.kt new file mode 100644 index 0000000..85093ed --- /dev/null +++ b/engine/src/main/kotlin/dev/m1sk9/lunaticChat/engine/protocol/PluginMessageChannel.kt @@ -0,0 +1,16 @@ +package dev.m1sk9.lunaticChat.engine.protocol + +/** + * The plugin messaging channel Paper and Velocity exchange [PluginMessage]s over. + * + * Both sides must agree on this exactly. Declaring it next to the codec keeps a rename from + * silently splitting the two halves of the protocol: a Paper server and a proxy that disagree + * still compile and start, they just stop talking. + */ +object PluginMessageChannel { + const val NAMESPACE = "lunaticchat" + const val NAME = "main" + + /** The channel in Bukkit's `namespace:name` form. */ + const val ID = "$NAMESPACE:$NAME" +} diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/CrossServerChatManager.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/CrossServerChatManager.kt index 53eced1..70ee6d4 100644 --- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/CrossServerChatManager.kt +++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/CrossServerChatManager.kt @@ -1,12 +1,13 @@ package dev.m1sk9.lunaticChat.paper.velocity import dev.m1sk9.lunaticChat.engine.protocol.PluginMessage +import dev.m1sk9.lunaticChat.engine.protocol.PluginMessageChannel +import dev.m1sk9.lunaticChat.engine.protocol.PluginMessageCodec import dev.m1sk9.lunaticChat.paper.config.LunaticChatConfiguration import net.kyori.adventure.text.Component import net.kyori.adventure.text.serializer.legacy.LegacyComponentSerializer import org.bukkit.plugin.Plugin import java.util.UUID -import java.util.concurrent.ConcurrentHashMap import java.util.logging.Level import java.util.logging.Logger @@ -24,15 +25,7 @@ class CrossServerChatManager( private val configuration: LunaticChatConfiguration, private val cacheSize: Int = 100, ) { - companion object { - private const val CLEANUP_THRESHOLD_MILLIS = 60_000L - } - - /** - * Cache of recently processed message IDs (messageId -> timestamp) - * Used for deduplication - */ - private val processedMessages = ConcurrentHashMap<String, Long>() + private val processedMessages = MessageDeduplicationCache(cacheSize, logger, "global chat") /** * Sends a global chat message to Velocity for cross-server broadcast @@ -51,7 +44,7 @@ class CrossServerChatManager( val serverName = configuration.features.velocityIntegration.serverName // Mark as processed immediately to prevent echo - processedMessages[messageId] = System.currentTimeMillis() + processedMessages.markProcessed(messageId) val globalChatMessage = PluginMessage.GlobalChatMessage( @@ -72,9 +65,8 @@ class CrossServerChatManager( if (player != null) { player.sendPluginMessage( plugin, - "lunaticchat:main", - dev.m1sk9.lunaticChat.engine.protocol.PluginMessageCodec - .encode(globalChatMessage), + PluginMessageChannel.ID, + PluginMessageCodec.encode(globalChatMessage), ) logger.info("Sent global chat message to Velocity: messageId=$messageId, player=$playerName") } else { @@ -85,11 +77,6 @@ class CrossServerChatManager( } }, ) - - // Cleanup old messages if cache is too large - if (processedMessages.size > cacheSize) { - cleanupOldMessages() - } } catch (e: Exception) { logger.log(Level.SEVERE, "Failed to send global chat message", e) } @@ -103,13 +90,11 @@ class CrossServerChatManager( fun handleIncomingMessage(message: PluginMessage.GlobalChatMessage) { try { // Check if already processed (deduplication) - if (!shouldProcessMessage(message.messageId)) { + if (!processedMessages.isNew(message.messageId)) { logger.fine("Ignoring duplicate message: messageId=${message.messageId}") return } - - // Mark as processed - processedMessages[message.messageId] = System.currentTimeMillis() + processedMessages.markProcessed(message.messageId) // Broadcast to all players on this server val formattedMessage = formatCrossServerMessage(message) @@ -127,11 +112,6 @@ class CrossServerChatManager( "Broadcasted global chat message from ${message.serverName}: " + "player=${message.playerName}, messageId=${message.messageId}", ) - - // Cleanup if needed - if (processedMessages.size > cacheSize) { - cleanupOldMessages() - } } catch (e: Exception) { logger.log(Level.SEVERE, "Failed to handle incoming global chat message", e) } @@ -153,54 +133,4 @@ class CrossServerChatManager( return LegacyComponentSerializer.legacySection().deserialize(formattedText) } - - /** - * Checks if a message should be processed (not a duplicate) - * - * @param messageId Message ID to check - * @return true if message should be processed, false if it's a duplicate - */ - private fun shouldProcessMessage(messageId: String): Boolean = !processedMessages.containsKey(messageId) - - /** - * Removes old messages from the cache (LRU cleanup) - * Keeps only the most recent messages - */ - private fun cleanupOldMessages() { - try { - val currentTime = System.currentTimeMillis() - val cutoffTime = currentTime - CLEANUP_THRESHOLD_MILLIS - - // Collect keys to remove (ConcurrentHashMap iterator doesn't support remove()) - val keysToRemove = mutableListOf<String>() - processedMessages.entries.forEach { entry -> - if (entry.value < cutoffTime) { - keysToRemove.add(entry.key) - } - } - - // Remove expired entries - keysToRemove.forEach { key -> - processedMessages.remove(key) - } - var removedCount = keysToRemove.size - - // If still over cache size, remove oldest entries - if (processedMessages.size > cacheSize) { - val sortedEntries = processedMessages.entries.sortedBy { it.value } - val toRemove = processedMessages.size - cacheSize - - sortedEntries.take(toRemove).forEach { entry -> - processedMessages.remove(entry.key) - removedCount++ - } - } - - if (removedCount > 0) { - logger.fine("Cleaned up $removedCount old messages from deduplication cache") - } - } catch (e: Exception) { - logger.log(Level.WARNING, "Failed to cleanup old messages", e) - } - } } diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/CrossServerDirectMessageManager.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/CrossServerDirectMessageManager.kt index b8390a7..b35f6a5 100644 --- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/CrossServerDirectMessageManager.kt +++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/CrossServerDirectMessageManager.kt @@ -1,6 +1,7 @@ package dev.m1sk9.lunaticChat.paper.velocity import dev.m1sk9.lunaticChat.engine.protocol.PluginMessage +import dev.m1sk9.lunaticChat.engine.protocol.PluginMessageChannel import dev.m1sk9.lunaticChat.engine.protocol.PluginMessageCodec import dev.m1sk9.lunaticChat.paper.chat.handler.DirectMessageHandler import dev.m1sk9.lunaticChat.paper.config.LunaticChatConfiguration @@ -9,7 +10,6 @@ import dev.m1sk9.lunaticChat.paper.i18n.MessageFormatter import org.bukkit.entity.Player import org.bukkit.plugin.Plugin import java.util.UUID -import java.util.concurrent.ConcurrentHashMap import java.util.logging.Level import java.util.logging.Logger @@ -30,12 +30,7 @@ class CrossServerDirectMessageManager( private val languageManager: LanguageManager, private val cacheSize: Int = 100, ) { - companion object { - private const val CHANNEL = "lunaticchat:main" - private const val CLEANUP_THRESHOLD_MILLIS = 60_000L - } - - private val processedMessages = ConcurrentHashMap<String, Long>() + private val processedMessages = MessageDeduplicationCache(cacheSize, logger, "direct message") /** * Sends a direct message to a player on another server through Velocity. @@ -52,7 +47,7 @@ class CrossServerDirectMessageManager( ) { try { val messageId = UUID.randomUUID().toString() - processedMessages[messageId] = System.currentTimeMillis() + processedMessages.markProcessed(messageId) val relayedMessage = directMessageHandler.handleOutgoingCrossServerMessage( @@ -73,15 +68,11 @@ class CrossServerDirectMessageManager( message = relayedMessage, ) - sender.sendPluginMessage(plugin, CHANNEL, PluginMessageCodec.encode(relay)) + sender.sendPluginMessage(plugin, PluginMessageChannel.ID, PluginMessageCodec.encode(relay)) logger.info( "Sent direct message to Velocity: messageId=$messageId, " + "target=$targetName@$targetServerName", ) - - if (processedMessages.size > cacheSize) { - cleanupOldMessages() - } } catch (e: Exception) { logger.log(Level.SEVERE, "Failed to send cross-server direct message", e) } @@ -92,11 +83,11 @@ class CrossServerDirectMessageManager( */ fun handleIncomingMessage(message: PluginMessage.DirectMessageRelay) { try { - if (!shouldProcessMessage(message.messageId)) { + if (!processedMessages.isNew(message.messageId)) { logger.fine("Ignoring duplicate direct message: messageId=${message.messageId}") return } - processedMessages[message.messageId] = System.currentTimeMillis() + processedMessages.markProcessed(message.messageId) plugin.server.scheduler.runTask( plugin, @@ -117,10 +108,6 @@ class CrossServerDirectMessageManager( ) }, ) - - if (processedMessages.size > cacheSize) { - cleanupOldMessages() - } } catch (e: Exception) { logger.log(Level.SEVERE, "Failed to handle incoming direct message", e) } @@ -153,33 +140,4 @@ class CrossServerDirectMessageManager( logger.log(Level.SEVERE, "Failed to handle direct message error", e) } } - - private fun shouldProcessMessage(messageId: String): Boolean = !processedMessages.containsKey(messageId) - - private fun cleanupOldMessages() { - try { - val cutoffTime = System.currentTimeMillis() - CLEANUP_THRESHOLD_MILLIS - - val keysToRemove = processedMessages.entries.filter { it.value < cutoffTime }.map { it.key } - keysToRemove.forEach { processedMessages.remove(it) } - var removedCount = keysToRemove.size - - if (processedMessages.size > cacheSize) { - val toRemove = processedMessages.size - cacheSize - processedMessages.entries - .sortedBy { it.value } - .take(toRemove) - .forEach { - processedMessages.remove(it.key) - removedCount++ - } - } - - if (removedCount > 0) { - logger.fine("Cleaned up $removedCount old messages from direct message dedup cache") - } - } catch (e: Exception) { - logger.log(Level.WARNING, "Failed to cleanup old direct messages", e) - } - } } diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/MessageDeduplicationCache.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/MessageDeduplicationCache.kt new file mode 100644 index 0000000..e804e03 --- /dev/null +++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/MessageDeduplicationCache.kt @@ -0,0 +1,70 @@ +package dev.m1sk9.lunaticChat.paper.velocity + +import java.util.concurrent.ConcurrentHashMap +import java.util.logging.Level +import java.util.logging.Logger + +/** + * Remembers recently seen message IDs so a message relayed back to its origin server is dropped + * instead of echoed. + * + * Entries expire after [CLEANUP_THRESHOLD_MILLIS]; if the cache is still over [cacheSize] after + * that, the oldest entries go too. Entries are ordered by millisecond timestamp, so a burst of + * more than [cacheSize] messages inside one millisecond evicts arbitrarily among them. + * + * @param cacheSize Soft upper bound on retained entries + * @param logger Where cleanup failures are reported + * @param description Names this cache in log output + */ +class MessageDeduplicationCache( + private val cacheSize: Int, + private val logger: Logger, + private val description: String, +) { + companion object { + private const val CLEANUP_THRESHOLD_MILLIS = 60_000L + } + + private val processedMessages = ConcurrentHashMap<String, Long>() + + /** + * Records [messageId] as seen, evicting stale entries when the cache outgrows [cacheSize]. + */ + fun markProcessed(messageId: String) { + processedMessages[messageId] = System.currentTimeMillis() + if (processedMessages.size > cacheSize) { + evict() + } + } + + /** + * Returns true when [messageId] has not been seen yet. + */ + fun isNew(messageId: String): Boolean = !processedMessages.containsKey(messageId) + + private fun evict() { + try { + val cutoffTime = System.currentTimeMillis() - CLEANUP_THRESHOLD_MILLIS + + val expired = processedMessages.entries.filter { it.value < cutoffTime }.map { it.key } + expired.forEach { processedMessages.remove(it) } + var removedCount = expired.size + + if (processedMessages.size > cacheSize) { + processedMessages.entries + .sortedBy { it.value } + .take(processedMessages.size - cacheSize) + .forEach { + processedMessages.remove(it.key) + removedCount++ + } + } + + if (removedCount > 0) { + logger.fine("Cleaned up $removedCount old messages from $description dedup cache") + } + } catch (e: Exception) { + logger.log(Level.WARNING, "Failed to clean up $description dedup cache", e) + } + } +} diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/VelocityConnectionManager.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/VelocityConnectionManager.kt index 1f4ccb4..bdedb6f 100644 --- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/VelocityConnectionManager.kt +++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/velocity/VelocityConnectionManager.kt @@ -1,6 +1,7 @@ package dev.m1sk9.lunaticChat.paper.velocity import dev.m1sk9.lunaticChat.engine.protocol.PluginMessage +import dev.m1sk9.lunaticChat.engine.protocol.PluginMessageChannel import dev.m1sk9.lunaticChat.engine.protocol.PluginMessageCodec import dev.m1sk9.lunaticChat.engine.protocol.ProtocolVersion import org.bukkit.entity.Player @@ -21,7 +22,7 @@ class VelocityConnectionManager( private var remotePlayerRegistry: RemotePlayerRegistry? = null, ) : PluginMessageListener { companion object { - private const val CHANNEL = "lunaticchat:main" + private val CHANNEL = PluginMessageChannel.ID private const val HANDSHAKE_TIMEOUT_SECONDS = 5L } diff --git a/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/velocity/MessageDeduplicationCacheTest.kt b/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/velocity/MessageDeduplicationCacheTest.kt new file mode 100644 index 0000000..316b455 --- /dev/null +++ b/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/velocity/MessageDeduplicationCacheTest.kt @@ -0,0 +1,44 @@ +package dev.m1sk9.lunaticChat.paper.velocity + +import io.mockk.mockk +import java.util.logging.Logger +import kotlin.test.Test +import kotlin.test.assertFalse +import kotlin.test.assertTrue + +class MessageDeduplicationCacheTest { + private fun cache(cacheSize: Int) = MessageDeduplicationCache(cacheSize, mockk<Logger>(relaxed = true), "test") + + @Test + fun `an unseen message id is new`() { + assertTrue(cache(10).isNew("m1")) + } + + @Test + fun `a recorded message id is no longer new`() { + val cache = cache(10) + + cache.markProcessed("m1") + + assertFalse(cache.isNew("m1")) + } + + @Test + fun `recording one id does not mask another`() { + val cache = cache(10) + + cache.markProcessed("m1") + + assertTrue(cache.isNew("m2")) + } + + @Test + fun `eviction keeps the cache from growing without bound`() { + val cache = cache(4) + + repeat(100) { cache.markProcessed("m$it") } + + val remembered = (0 until 100).count { !cache.isNew("m$it") } + assertTrue(remembered <= 4, "expected at most 4 retained entries, got $remembered") + } +} diff --git a/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/CrossServerChatRelay.kt b/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/CrossServerChatRelay.kt index bedc302..f47db70 100644 --- a/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/CrossServerChatRelay.kt +++ b/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/CrossServerChatRelay.kt @@ -4,6 +4,7 @@ import com.velocitypowered.api.proxy.ProxyServer import com.velocitypowered.api.proxy.messages.MinecraftChannelIdentifier import com.velocitypowered.api.proxy.server.RegisteredServer import dev.m1sk9.lunaticChat.engine.protocol.PluginMessage +import dev.m1sk9.lunaticChat.engine.protocol.PluginMessageChannel import dev.m1sk9.lunaticChat.engine.protocol.PluginMessageCodec import org.slf4j.Logger @@ -18,7 +19,7 @@ class CrossServerChatRelay( private val logger: Logger, ) { companion object { - private val CHANNEL = MinecraftChannelIdentifier.create("lunaticchat", "main") + private val CHANNEL = MinecraftChannelIdentifier.create(PluginMessageChannel.NAMESPACE, PluginMessageChannel.NAME) } /** diff --git a/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/CrossServerDirectMessageRelay.kt b/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/CrossServerDirectMessageRelay.kt index 89cc15f..3746395 100644 --- a/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/CrossServerDirectMessageRelay.kt +++ b/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/CrossServerDirectMessageRelay.kt @@ -4,6 +4,7 @@ import com.velocitypowered.api.proxy.ProxyServer import com.velocitypowered.api.proxy.messages.MinecraftChannelIdentifier import com.velocitypowered.api.proxy.server.RegisteredServer import dev.m1sk9.lunaticChat.engine.protocol.PluginMessage +import dev.m1sk9.lunaticChat.engine.protocol.PluginMessageChannel import dev.m1sk9.lunaticChat.engine.protocol.PluginMessageCodec import org.slf4j.Logger @@ -19,7 +20,7 @@ class CrossServerDirectMessageRelay( private val logger: Logger, ) { companion object { - private val CHANNEL = MinecraftChannelIdentifier.create("lunaticchat", "main") + private val CHANNEL = MinecraftChannelIdentifier.create(PluginMessageChannel.NAMESPACE, PluginMessageChannel.NAME) } /** diff --git a/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/PluginMessageHandler.kt b/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/PluginMessageHandler.kt index 4472d1c..6b40a7e 100644 --- a/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/PluginMessageHandler.kt +++ b/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/messaging/PluginMessageHandler.kt @@ -6,6 +6,7 @@ import com.velocitypowered.api.proxy.ProxyServer import com.velocitypowered.api.proxy.ServerConnection import com.velocitypowered.api.proxy.messages.MinecraftChannelIdentifier import dev.m1sk9.lunaticChat.engine.protocol.PluginMessage +import dev.m1sk9.lunaticChat.engine.protocol.PluginMessageChannel import dev.m1sk9.lunaticChat.engine.protocol.PluginMessageCodec import dev.m1sk9.lunaticChat.engine.protocol.ProtocolVersion import dev.m1sk9.lunaticChat.velocity.presence.PresenceTracker @@ -30,7 +31,7 @@ class PluginMessageHandler( private val presenceTracker: PresenceTracker, ) { companion object { - private val CHANNEL = MinecraftChannelIdentifier.create("lunaticchat", "main") + private val CHANNEL = MinecraftChannelIdentifier.create(PluginMessageChannel.NAMESPACE, PluginMessageChannel.NAME) } /** diff --git a/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/presence/PresenceTracker.kt b/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/presence/PresenceTracker.kt index f9442ee..59dbb5b 100644 --- a/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/presence/PresenceTracker.kt +++ b/platform-velocity/src/main/kotlin/dev/m1sk9/lunaticChat/velocity/presence/PresenceTracker.kt @@ -8,6 +8,7 @@ import com.velocitypowered.api.proxy.ProxyServer import com.velocitypowered.api.proxy.messages.MinecraftChannelIdentifier import com.velocitypowered.api.proxy.server.RegisteredServer import dev.m1sk9.lunaticChat.engine.protocol.PluginMessage +import dev.m1sk9.lunaticChat.engine.protocol.PluginMessageChannel import dev.m1sk9.lunaticChat.engine.protocol.PluginMessageCodec import dev.m1sk9.lunaticChat.engine.protocol.PresenceEntry import org.slf4j.Logger @@ -28,7 +29,7 @@ class PresenceTracker( private val logger: Logger, ) { companion object { - private val CHANNEL = MinecraftChannelIdentifier.create("lunaticchat", "main") + private val CHANNEL = MinecraftChannelIdentifier.create(PluginMessageChannel.NAMESPACE, PluginMessageChannel.NAME) } /** |