refactor: bound the delivery queue and drop a departed player's backlog

An item takes up to the conversion timeout to drain, far slower than a player
can send, so an unbounded queue grew for as long as a macro ran - each item
holding its sender and recipient alive and arriving minutes after it was typed.
Refusing the overflow is at least visible to the player.

Releasing a player now cancels their worker instead of letting the backlog run.
Finishing it would spend a round trip per item writing to somebody who has left,
and keep both players reachable until it drained.

Chosen over DROP_LATEST on the channel: that reports success to trySend and
drops silently, which would make the warning about discarded work unreachable in
the case it was written for.

Co-Authored-By: Claude <noreply@anthropic.com>

1 files changed, 56 insertions, 26 deletions

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 f5e4a28..5973d64 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
@@ -2,6 +2,7 @@ package dev.m1sk9.lunaticChat.paper
 
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.SendChannel
 import kotlinx.coroutines.isActive
@@ -26,10 +27,27 @@ class PerPlayerWorkQueue(
     private val scope: CoroutineScope,
     private val logger: Logger,
 ) {
-    private val queues = ConcurrentHashMap<UUID, SendChannel<suspend () -> Unit>>()
+    private val queues = ConcurrentHashMap<UUID, Worker>()
+
+    private companion object {
+        /**
+         * Bounded so a player cannot make the server hold work it will never get through.
+         *
+         * An item takes up to the conversion timeout to drain, which is far slower than a player can
+         * send; an unbounded queue therefore grew for as long as a macro ran, each item holding its
+         * sender and recipient alive and arriving minutes after it was typed. Refusing the overflow
+         * is visible to the player, unlike delivering it late.
+         */
+        const val CAPACITY = 8
+    }
+
+    private class Worker(
+        val channel: SendChannel<suspend () -> Unit>,
+        val job: Job,
+    )
 
     /**
-     * Queues [work] behind anything already pending for [playerId].
+     * Queues [work] behind anything already pending for [playerId], unless their queue is full.
      */
     fun submit(
         playerId: UUID,
@@ -38,40 +56,52 @@ class PerPlayerWorkQueue(
         // Checked rather than left to trySend: cancelling the scope kills the worker coroutines but
         // does not close their channels, so after shutdown trySend would keep reporting success for
         // work nothing will ever read.
-        val accepted = scope.isActive && queues.computeIfAbsent(playerId) { startWorker() }.trySend(work).isSuccess
+        val accepted =
+            scope.isActive &&
+                queues
+                    .computeIfAbsent(playerId) { startWorker() }
+                    .channel
+                    .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")
+            // Reachable when the player is sending faster than delivery drains, once the scope is
+            // cancelled at shutdown, or if their 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 full or closed")
         }
     }
 
     /**
-     * Drops the player's queue once they can no longer send anything. Work already queued still
-     * runs; without this the map and its worker coroutines would grow for the life of the server.
+     * Drops the player's queue once they can no longer send anything.
+     *
+     * Anything still pending is abandoned rather than delivered: it was addressed to or sent by a
+     * player who has left, so finishing it would spend a conversion round trip per item to write to
+     * nobody, while holding both players alive until the backlog drained.
      */
     fun release(playerId: UUID) {
-        queues.remove(playerId)?.close()
+        queues.remove(playerId)?.let {
+            it.channel.close()
+            it.job.cancel()
+        }
     }
 
-    private fun startWorker(): SendChannel<suspend () -> Unit> {
-        // Unlimited so that submit never suspends or drops work on the caller's thread.
-        val channel = Channel<suspend () -> Unit>(Channel.UNLIMITED)
-        scope.launch {
-            for (work in channel) {
-                // 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)
+    private fun startWorker(): Worker {
+        val channel = Channel<suspend () -> Unit>(CAPACITY)
+        val job =
+            scope.launch {
+                for (work in channel) {
+                    // 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
+        return Worker(channel, job)
     }
 }