polylan/polylan-submitter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(clock): dedup tasks lock bug removal

Browse Source
This commit is contained in:
Legrems
2026-10-16 01:08:50 +02:00
parent 27d3c08a72
commit 1ca4643774
5 changed files with 113 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
polylan_submitter/clock/tasks.py
+17 -7
View File
@@ -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)
polylan_submitter/clock/tests.py
+57
View File
@@ -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))
polylan_submitter/games/admin.py
+1 -1
View File
@@ -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)."
)
polylan_submitter/games/tasks.py
+18 -9
View File
@@ -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)
polylan_submitter/games/tests.py
+20 -5
View File
@@ -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()