From 1ca464377461bef72600bbd802c01726c60603d2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lo=C3=AFc=20Gremaud?= Date: Fri, 16 Oct 2026 01:08:50 +0200 Subject: [PATCH] fix(clock): dedup tasks lock bug removal --- polylan_submitter/clock/tasks.py | 24 ++++++++++---- polylan_submitter/clock/tests.py | 57 ++++++++++++++++++++++++++++++++ polylan_submitter/games/admin.py | 2 +- polylan_submitter/games/tasks.py | 27 ++++++++++----- polylan_submitter/games/tests.py | 25 +++++++++++--- 5 files changed, 113 insertions(+), 22 deletions(-) create mode 100644 polylan_submitter/clock/tests.py diff --git a/polylan_submitter/clock/tasks.py b/polylan_submitter/clock/tasks.py index 394da9a..86716c2 100644 --- a/polylan_submitter/clock/tasks.py +++ b/polylan_submitter/clock/tasks.py @@ -12,6 +12,7 @@ acts. """ from datetime import timedelta +from uuid import uuid4 from asgiref.sync import async_to_sync from channels.layers import get_channel_layer @@ -30,13 +31,20 @@ HEARTBEAT_INTERVAL_SECONDS = 1 # One chain ticks every event with the game enabled - this lock keeps a # second chain from starting underneath an already-running one (e.g. two # EventGame saves close together). Refreshed every tick, wide enough that a -# single slow tick doesn't let a second chain slip in. +# single slow tick doesn't let a second chain slip in. The lock holds the +# chain id: if it expires anyway (worker down/restarted for a few seconds) and +# a new chain starts, the old one stops at its next tick (instead of both +# ticking forever). HEARTBEAT_LOCK_KEY = "clock:heartbeat:lock" HEARTBEAT_LOCK_TIMEOUT_SECONDS = HEARTBEAT_INTERVAL_SECONDS * 5 @task -def clock_heartbeat_task(): +def clock_heartbeat_task(chain_id: str): + if cache.get(HEARTBEAT_LOCK_KEY) != chain_id: + # Another chain owns the lock (or it expired), stop this one + return + event_games = list( EventGame.objects.filter(game__path="clock", enabled=True).select_related( "event" @@ -46,7 +54,8 @@ def clock_heartbeat_task(): if not event_games: # Nothing left to tick - let the chain die. start_clock_heartbeat() # starts a fresh one next time an event enables the game. - cache.delete(HEARTBEAT_LOCK_KEY) + if cache.get(HEARTBEAT_LOCK_KEY) == chain_id: + cache.delete(HEARTBEAT_LOCK_KEY) return channel_layer = get_channel_layer() @@ -67,13 +76,14 @@ def clock_heartbeat_task(): }, ) - cache.set(HEARTBEAT_LOCK_KEY, 1, timeout=HEARTBEAT_LOCK_TIMEOUT_SECONDS) + cache.set(HEARTBEAT_LOCK_KEY, chain_id, timeout=HEARTBEAT_LOCK_TIMEOUT_SECONDS) clock_heartbeat_task.using( run_after=timezone.now() + timedelta(seconds=HEARTBEAT_INTERVAL_SECONDS) - ).enqueue() + ).enqueue(chain_id) def start_clock_heartbeat(): """Enqueues the heartbeat chain, unless one is already running.""" - if cache.add(HEARTBEAT_LOCK_KEY, 1, timeout=HEARTBEAT_LOCK_TIMEOUT_SECONDS): - clock_heartbeat_task.enqueue() + chain_id = uuid4().hex + if cache.add(HEARTBEAT_LOCK_KEY, chain_id, timeout=HEARTBEAT_LOCK_TIMEOUT_SECONDS): + clock_heartbeat_task.enqueue(chain_id) diff --git a/polylan_submitter/clock/tests.py b/polylan_submitter/clock/tests.py new file mode 100644 index 0000000..ade03ee --- /dev/null +++ b/polylan_submitter/clock/tests.py @@ -0,0 +1,57 @@ +from unittest import mock + +from django.core.cache import cache +from django.test import TestCase, override_settings + +from events.models import Event +from games.models import EventGame, Game + +from .tasks import HEARTBEAT_LOCK_KEY, clock_heartbeat_task, start_clock_heartbeat + +LOCMEM_CACHE = {"default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}} + + +@override_settings(CACHES=LOCMEM_CACHE) +@mock.patch("clock.tasks.get_channel_layer") +class ClockHeartbeatTests(TestCase): + def setUp(self): + cache.clear() + event = Event.objects.create(name="Event", path="event") + clock = Game.objects.create(name="Clock", path="clock", custom_path=True) + self.event_game = EventGame.objects.create(event=event, game=clock) + cache.clear() # post_save signal can start a chain + + @mock.patch("clock.tasks.clock_heartbeat_task") + def test_only_one_chain_is_started(self, task, _): + start_clock_heartbeat() + start_clock_heartbeat() + + task.enqueue.assert_called_once_with(cache.get(HEARTBEAT_LOCK_KEY)) + + def test_owning_chain_reschedules_itself(self, channel_layer): + channel_layer.return_value.group_send = mock.AsyncMock() + cache.set(HEARTBEAT_LOCK_KEY, "chain") + with mock.patch("clock.tasks.clock_heartbeat_task") as patched: + clock_heartbeat_task.call("chain") + + patched.using.return_value.enqueue.assert_called_once_with("chain") + channel_layer.return_value.group_send.assert_awaited_once() + self.assertEqual(cache.get(HEARTBEAT_LOCK_KEY), "chain") + + def test_superseded_chain_stops(self, channel_layer): + cache.set(HEARTBEAT_LOCK_KEY, "new-chain") + with mock.patch("clock.tasks.clock_heartbeat_task") as patched: + clock_heartbeat_task.call("old-chain") + + patched.using.assert_not_called() + channel_layer.assert_not_called() + self.assertEqual(cache.get(HEARTBEAT_LOCK_KEY), "new-chain") + + def test_chain_dies_and_releases_lock_when_game_disabled(self, _): + EventGame.objects.filter(pk=self.event_game.pk).update(enabled=False) + cache.set(HEARTBEAT_LOCK_KEY, "chain") + with mock.patch("clock.tasks.clock_heartbeat_task") as patched: + clock_heartbeat_task.call("chain") + + patched.using.assert_not_called() + self.assertIsNone(cache.get(HEARTBEAT_LOCK_KEY)) diff --git a/polylan_submitter/games/admin.py b/polylan_submitter/games/admin.py index 1d0cd59..be65b06 100644 --- a/polylan_submitter/games/admin.py +++ b/polylan_submitter/games/admin.py @@ -40,7 +40,7 @@ class GameAdmin(admin.ModelAdmin): game_ids = list( queryset.filter(steam_app_id__isnull=False).values_list("id", flat=True) ) - refresh_steam_games_task.enqueue(game_ids=game_ids, reschedule=False) + refresh_steam_games_task.enqueue(game_ids=game_ids) self.message_user( request, f"Queued Steam store refresh for {len(game_ids)} game(s)." ) diff --git a/polylan_submitter/games/tasks.py b/polylan_submitter/games/tasks.py index 01ba3dd..7ac1af7 100644 --- a/polylan_submitter/games/tasks.py +++ b/polylan_submitter/games/tasks.py @@ -1,4 +1,5 @@ from datetime import timedelta +from uuid import uuid4 from django.conf import settings from django.core.cache import cache @@ -10,7 +11,9 @@ from .services.steam_store import refresh_steam_games # Set while a periodic refresh is queued, so there's only ever one loop running, # whatever the number of workers/containers trying to start it. It expires on # its own if the loop dies (worker killed mid-job, Redis queue flushed, ...), -# letting the next schedule_steam_refresh() start a new one. +# letting the next schedule_steam_refresh() start a new one. The guard holds the +# loop id, so if it expires while the old loop is just late, the old one stops +# at its next run (instead of 2 loops running side by side). STEAM_REFRESH_GUARD_KEY = "games:steam_refresh:scheduled" @@ -25,28 +28,34 @@ def _guard_timeout() -> int: def schedule_steam_refresh() -> bool: """Start the periodic refresh loop right away, unless one is already queued. Safe to call from anywhere, any number of times.""" - if not cache.add(STEAM_REFRESH_GUARD_KEY, 1, timeout=_guard_timeout()): + loop_id = uuid4().hex + if not cache.add(STEAM_REFRESH_GUARD_KEY, loop_id, timeout=_guard_timeout()): return False - refresh_steam_games_task.enqueue() + refresh_steam_games_task.enqueue(loop_id=loop_id) return True @task def refresh_steam_games_task( - game_ids: list[int] | None = None, reschedule: bool = True + game_ids: list[int] | None = None, loop_id: str | None = None ): """Refresh Steam store info, then re-enqueue itself every STEAM_STORE_REFRESH_MINUTES (same self-rescheduling idea as sync_event_game_scopes_task, no external scheduler needed). - reschedule=False is a one-off run (e.g. the admin action) that leaves the - periodic loop alone.""" + Without loop_id it's a one-off run (e.g. the admin action) that leaves the + periodic loop alone. With loop_id, it only continues while the guard still + holds this loop_id.""" + if loop_id is not None and cache.get(STEAM_REFRESH_GUARD_KEY) != loop_id: + # Another loop owns the guard (or it expired), stop this one + return + try: refresh_steam_games(game_ids) finally: - if reschedule: - cache.set(STEAM_REFRESH_GUARD_KEY, 1, timeout=_guard_timeout()) + if loop_id is not None: + cache.set(STEAM_REFRESH_GUARD_KEY, loop_id, timeout=_guard_timeout()) refresh_steam_games_task.using( run_after=timezone.now() + steam_refresh_interval() - ).enqueue() + ).enqueue(loop_id=loop_id) diff --git a/polylan_submitter/games/tests.py b/polylan_submitter/games/tests.py index 8142214..9e6f4bd 100644 --- a/polylan_submitter/games/tests.py +++ b/polylan_submitter/games/tests.py @@ -9,7 +9,11 @@ from events.models import Event from .models import EventGame, Game from .services.steam_store import refresh_steam_games -from .tasks import refresh_steam_games_task, schedule_steam_refresh +from .tasks import ( + STEAM_REFRESH_GUARD_KEY, + refresh_steam_games_task, + schedule_steam_refresh, +) LOCMEM_CACHE = {"default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}} @@ -83,21 +87,32 @@ class ScheduleSteamRefreshTests(TestCase): def test_only_one_loop_is_scheduled(self, task): self.assertTrue(schedule_steam_refresh()) self.assertFalse(schedule_steam_refresh()) - task.enqueue.assert_called_once_with() + task.enqueue.assert_called_once_with(loop_id=cache.get(STEAM_REFRESH_GUARD_KEY)) @mock.patch("games.tasks.refresh_steam_games") def test_task_reschedules_itself_even_on_failure(self, refresh): + cache.set(STEAM_REFRESH_GUARD_KEY, "loop") refresh.side_effect = RuntimeError("boom") with mock.patch("games.tasks.refresh_steam_games_task") as patched: with self.assertRaises(RuntimeError): - refresh_steam_games_task.call() + refresh_steam_games_task.call(loop_id="loop") - patched.using.return_value.enqueue.assert_called_once_with() + patched.using.return_value.enqueue.assert_called_once_with(loop_id="loop") + + @mock.patch("games.tasks.refresh_steam_games") + def test_superseded_loop_stops(self, refresh): + cache.set(STEAM_REFRESH_GUARD_KEY, "new-loop") + with mock.patch("games.tasks.refresh_steam_games_task") as patched: + refresh_steam_games_task.call(loop_id="old-loop") + + refresh.assert_not_called() + patched.using.assert_not_called() + self.assertEqual(cache.get(STEAM_REFRESH_GUARD_KEY), "new-loop") @mock.patch("games.tasks.refresh_steam_games") def test_one_off_run_does_not_reschedule(self, refresh): with mock.patch("games.tasks.refresh_steam_games_task") as patched: - refresh_steam_games_task.call(game_ids=[1], reschedule=False) + refresh_steam_games_task.call(game_ids=[1]) refresh.assert_called_once_with([1]) patched.using.assert_not_called()