Drive timeouts from a shared Redis deadline queue

Turn auto-play and hand-end auto-continue were process-local asyncio
tasks armed only by client connects and state broadcasts: with no
sockets connected the next turn's timer was never armed, a hand-end
timer died with its worker, and neither survived a pod restart.

Deadlines are now driven by the absolute timestamps persisted on the
game state and enqueued in a shared Redis sorted set. Every worker runs
a consumer that fires due entries under the per-game lock after
revalidating them against the live state, so timeouts no longer depend
on any player being connected and survive the death of any worker.
Delivery is at-least-once: entries are removed only after processing,
and revalidation makes duplicate deliveries no-ops.

Queue entries carry the deadline as integer epoch milliseconds, which
also serves as the revalidation token, and the score derives from the
same value.

server/tests/test_store.py

@@ -108,6 +108,29 @@ class InMemoryGameStoreTest(unittest.TestCase):
             ["holder-enter", "holder-exit", "contender"], order
         )
 
+    @async_test
+    async def test_deadline_queue(self) -> None:
+        store = InMemoryGameStore()
+        self.assertIsNone(await store.next_deadline())
+        self.assertEqual([], await store.due_deadlines(now=100.0))
+
+        await store.add_deadline("b", due_at=50.0)
+        await store.add_deadline("a", due_at=10.0)
+        await store.add_deadline("c", due_at=200.0)
+        # Re-adding an existing member only updates its due time.
+        await store.add_deadline("b", due_at=60.0)
+
+        self.assertEqual(10.0, await store.next_deadline())
+        self.assertEqual(["a"], await store.due_deadlines(now=10.0))
+        self.assertEqual(["a", "b"], await store.due_deadlines(now=100.0))
+        # Due entries come out in due-time order and stay queued until removed.
+        self.assertEqual(["a", "b"], await store.due_deadlines(now=100.0))
+
+        await store.remove_deadline("a")
+        await store.remove_deadline("a")  # removing twice is a no-op
+        self.assertEqual(60.0, await store.next_deadline())
+        self.assertEqual(["b"], await store.due_deadlines(now=100.0))
+
 
 if __name__ == "__main__":
     unittest.main()