mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-25 11:04:07 +00:00
SergeantPanda
GitHub
commit
67ccbd70bc
4 changed files with 364 additions and 31 deletions
|
|
@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
|||
|
||||
## [Unreleased]
|
||||
|
||||
### Fixed
|
||||
|
||||
- **Fresh AIO installs no longer crash during first-boot migrate when Redis is not up yet.** In AIO, `manage.py migrate` runs in the entrypoint before uWSGI starts Redis via `attach-daemon`. After CoreSettings group caching landed in 0.28.0, data migrations such as `m3u.0003` called live `CoreSettings.get_default_user_agent_id()` → Redis and failed with connection refused, leaving a partially migrated DB and a boot loop. Settings group cache reads/writes now fall back to Postgres on connectivity/timeout errors (with a throttled warning), skip cache fill on backend failure so a flapping Redis cannot re-poison entries after invalidate, and leave auth/protocol Redis errors loud. A regression test migrates a throwaway empty database with Redis unreachable. (Fixes #1459)
|
||||
|
||||
## [0.28.0] - 2026-07-19
|
||||
|
||||
### Added
|
||||
|
|
|
|||
162
core/models.py
162
core/models.py
|
|
@ -1,11 +1,21 @@
|
|||
# core/models.py
|
||||
|
||||
import logging
|
||||
import time
|
||||
from shlex import split as shlex_split
|
||||
|
||||
from django.conf import settings
|
||||
from django.core.cache import cache
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.db import models
|
||||
from django.utils.text import slugify
|
||||
from django.core.exceptions import ValidationError
|
||||
from django_redis.exceptions import ConnectionInterrupted
|
||||
from redis.exceptions import AuthenticationError as RedisAuthenticationError
|
||||
from redis.exceptions import AuthorizationError as RedisAuthorizationError
|
||||
from redis.exceptions import ConnectionError as RedisConnectionError
|
||||
from redis.exceptions import TimeoutError as RedisTimeoutError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class UserAgent(models.Model):
|
||||
|
|
@ -204,6 +214,42 @@ _GROUP_CACHE_PREFIX = "coresettings:group:"
|
|||
_GROUP_CACHE_VER_PREFIX = "coresettings:groupver:"
|
||||
_GROUP_CACHE_TTL_SECONDS = 300
|
||||
|
||||
# Connectivity / timeout only. ResponseError (WRONGTYPE) and similar must still
|
||||
# propagate. Note: redis-py's AuthenticationError / AuthorizationError subclass
|
||||
# ConnectionError, so helpers re-raise those after the catch.
|
||||
_GROUP_CACHE_BACKEND_ERRORS = (
|
||||
RedisConnectionError,
|
||||
RedisTimeoutError,
|
||||
ConnectionInterrupted,
|
||||
OSError,
|
||||
TimeoutError,
|
||||
)
|
||||
_GROUP_CACHE_RERAISE_ERRORS = (RedisAuthenticationError, RedisAuthorizationError)
|
||||
|
||||
# Distinct from a normal cache miss (None) so version-guard compares never
|
||||
# collapse to None == None after a backend failure.
|
||||
_CACHE_BACKEND_ERROR = object()
|
||||
|
||||
_GROUP_CACHE_ERROR_LOG_INTERVAL_SECONDS = 60
|
||||
_last_group_cache_error_log_at = 0.0
|
||||
|
||||
|
||||
def _log_group_cache_backend_error(operation, key, exc):
|
||||
"""Warn when settings cache degrades to Postgres (throttled)."""
|
||||
global _last_group_cache_error_log_at
|
||||
|
||||
now = time.time()
|
||||
if now - _last_group_cache_error_log_at < _GROUP_CACHE_ERROR_LOG_INTERVAL_SECONDS:
|
||||
return
|
||||
_last_group_cache_error_log_at = now
|
||||
logger.warning(
|
||||
"CoreSettings group cache %s failed for %s (%s: %s); falling back to Postgres",
|
||||
operation,
|
||||
key,
|
||||
type(exc).__name__,
|
||||
exc,
|
||||
)
|
||||
|
||||
|
||||
class CoreSettings(models.Model):
|
||||
key = models.CharField(
|
||||
|
|
@ -229,17 +275,55 @@ class CoreSettings(models.Model):
|
|||
def group_cache_ver_key(cls, key):
|
||||
return f"{_GROUP_CACHE_VER_PREFIX}{key}"
|
||||
|
||||
@classmethod
|
||||
def _cache_get(cls, key, default=None):
|
||||
"""Read from Django cache.
|
||||
|
||||
Returns ``_CACHE_BACKEND_ERROR`` if Redis is unreachable so callers can
|
||||
distinguish that from a normal miss. AIO starts Redis via uWSGI after
|
||||
``migrate``, so settings reads during data migrations must not
|
||||
hard-require Redis. Local connection refused fails immediately (no
|
||||
connect-timeout wait).
|
||||
"""
|
||||
try:
|
||||
return cache.get(key, default)
|
||||
except _GROUP_CACHE_RERAISE_ERRORS:
|
||||
raise
|
||||
except _GROUP_CACHE_BACKEND_ERRORS as exc:
|
||||
_log_group_cache_backend_error("get", key, exc)
|
||||
return _CACHE_BACKEND_ERROR
|
||||
|
||||
@classmethod
|
||||
def _cache_set(cls, key, value, timeout=None):
|
||||
"""Write to Django cache; no-op if Redis is unreachable."""
|
||||
try:
|
||||
cache.set(key, value, timeout=timeout)
|
||||
return True
|
||||
except _GROUP_CACHE_RERAISE_ERRORS:
|
||||
raise
|
||||
except _GROUP_CACHE_BACKEND_ERRORS as exc:
|
||||
_log_group_cache_backend_error("set", key, exc)
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
def _cache_delete(cls, key):
|
||||
"""Delete from Django cache; no-op if Redis is unreachable."""
|
||||
try:
|
||||
cache.delete(key)
|
||||
return True
|
||||
except _GROUP_CACHE_RERAISE_ERRORS:
|
||||
raise
|
||||
except _GROUP_CACHE_BACKEND_ERRORS as exc:
|
||||
_log_group_cache_backend_error("delete", key, exc)
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
def invalidate_group_cache(cls, key):
|
||||
"""Drop the cached JSON for a settings group (all workers share Redis)."""
|
||||
import time
|
||||
|
||||
from django.core.cache import cache
|
||||
|
||||
cache.delete(cls.group_cache_key(key))
|
||||
cls._cache_delete(cls.group_cache_key(key))
|
||||
# Monotonic bump so in-flight _get_group fills skip cache.set.
|
||||
# timeout=None: never expire (version must outlive group entries).
|
||||
cache.set(cls.group_cache_ver_key(key), time.time_ns(), timeout=None)
|
||||
cls._cache_set(cls.group_cache_ver_key(key), time.time_ns(), timeout=None)
|
||||
if key == PROXY_SETTINGS_KEY:
|
||||
# Proxy workers also keep a short process-local copy.
|
||||
try:
|
||||
|
|
@ -249,40 +333,56 @@ class CoreSettings(models.Model):
|
|||
except Exception:
|
||||
pass
|
||||
|
||||
# Helper methods to get/set grouped settings
|
||||
@classmethod
|
||||
def _get_group(cls, key, defaults=None):
|
||||
"""Get a settings group, returning defaults if not found.
|
||||
|
||||
Results are cached in Redis so hot paths (proxy, XC, catchup) do not
|
||||
hit Postgres on every client request. Mutations go through ``save`` /
|
||||
``_update_group``, which invalidate via CoreSettings post_save /
|
||||
post_delete signals.
|
||||
"""
|
||||
import copy
|
||||
|
||||
from django.core.cache import cache
|
||||
|
||||
defaults = defaults or {}
|
||||
cache_key = cls.group_cache_key(key)
|
||||
ver_key = cls.group_cache_ver_key(key)
|
||||
cached = cache.get(cache_key)
|
||||
if isinstance(cached, dict):
|
||||
return copy.deepcopy(cached)
|
||||
|
||||
ver_before = cache.get(ver_key)
|
||||
def _load_group_value(cls, key, defaults):
|
||||
"""Read a settings group from Postgres (no cache)."""
|
||||
try:
|
||||
value = cls.objects.get(key=key).value or defaults
|
||||
if not isinstance(value, dict):
|
||||
value = defaults
|
||||
except cls.DoesNotExist:
|
||||
value = defaults
|
||||
return value
|
||||
|
||||
# Helper methods to get/set grouped settings
|
||||
@classmethod
|
||||
def _get_group(cls, key, defaults=None):
|
||||
"""Get a settings group, returning defaults if not found.
|
||||
|
||||
Results are cached in Redis so hot paths (proxy, XC, catchup) do not
|
||||
hit Postgres on every client request. If Redis is down (for example
|
||||
during AIO first-boot migrate), reads fall through to Postgres and
|
||||
skip cache fill so a flapping backend cannot re-poison Redis after
|
||||
invalidate. Mutations go through ``save`` / ``_update_group``, which
|
||||
invalidate via CoreSettings post_save / post_delete signals.
|
||||
"""
|
||||
import copy
|
||||
|
||||
defaults = defaults or {}
|
||||
cache_key = cls.group_cache_key(key)
|
||||
ver_key = cls.group_cache_ver_key(key)
|
||||
cached = cls._cache_get(cache_key)
|
||||
if isinstance(cached, dict):
|
||||
return copy.deepcopy(cached)
|
||||
|
||||
# Backend errors are not normal misses: read DB and skip fill so a
|
||||
# flapping backend cannot collapse the version guard to None == None.
|
||||
if cached is _CACHE_BACKEND_ERROR:
|
||||
return copy.deepcopy(cls._load_group_value(key, defaults))
|
||||
|
||||
ver_before = cls._cache_get(ver_key)
|
||||
if ver_before is _CACHE_BACKEND_ERROR:
|
||||
return copy.deepcopy(cls._load_group_value(key, defaults))
|
||||
|
||||
value = copy.deepcopy(cls._load_group_value(key, defaults))
|
||||
|
||||
value = copy.deepcopy(value)
|
||||
# Skip fill if an invalidate landed during the DB read (avoids
|
||||
# re-caching a stale snapshot for the full TTL).
|
||||
if cache.get(ver_key) == ver_before:
|
||||
cache.set(cache_key, value, timeout=_GROUP_CACHE_TTL_SECONDS)
|
||||
ver_after = cls._cache_get(ver_key)
|
||||
if ver_after is _CACHE_BACKEND_ERROR or ver_after != ver_before:
|
||||
return value
|
||||
|
||||
cls._cache_set(cache_key, value, timeout=_GROUP_CACHE_TTL_SECONDS)
|
||||
return copy.deepcopy(value)
|
||||
|
||||
@classmethod
|
||||
|
|
|
|||
|
|
@ -4,11 +4,14 @@ from django.core.cache import cache
|
|||
from django.test import TestCase, SimpleTestCase
|
||||
|
||||
from apps.epg.models import EPGSource, EPGSourceIndex
|
||||
import core.models as core_models
|
||||
from core.models import (
|
||||
CoreSettings,
|
||||
DVR_SETTINGS_KEY,
|
||||
EPG_SETTINGS_KEY,
|
||||
STREAM_SETTINGS_KEY,
|
||||
SYSTEM_SETTINGS_KEY,
|
||||
_CACHE_BACKEND_ERROR,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -18,6 +21,8 @@ class CoreSettingsGroupCacheTests(TestCase):
|
|||
def setUp(self):
|
||||
cache.clear()
|
||||
CoreSettings.objects.filter(key=SYSTEM_SETTINGS_KEY).delete()
|
||||
# Allow fallback warnings to emit in each test.
|
||||
core_models._last_group_cache_error_log_at = 0.0
|
||||
|
||||
def tearDown(self):
|
||||
# DB rollback does not undo Redis entries written during the test.
|
||||
|
|
@ -140,6 +145,115 @@ class CoreSettingsGroupCacheTests(TestCase):
|
|||
with self.assertNumQueries(0):
|
||||
self.assertTrue(network_access_allowed(request, "STREAMS"))
|
||||
|
||||
def test_get_group_falls_back_to_db_when_redis_unavailable(self):
|
||||
"""AIO migrate runs before Redis; settings reads must use Postgres.
|
||||
|
||||
Fresh AIO installs run ``manage.py migrate`` before uWSGI starts
|
||||
Redis. Data migrations such as m3u.0003 call
|
||||
``CoreSettings.get_default_user_agent_id()``, which must not raise
|
||||
when the cache backend is unreachable.
|
||||
"""
|
||||
from redis.exceptions import ConnectionError as RedisConnectionError
|
||||
|
||||
CoreSettings.objects.update_or_create(
|
||||
key=STREAM_SETTINGS_KEY,
|
||||
defaults={
|
||||
"name": "Stream Settings",
|
||||
"value": {"default_user_agent": "ua-from-db"},
|
||||
},
|
||||
)
|
||||
redis_down = RedisConnectionError(
|
||||
"Error 111 connecting to localhost:6379"
|
||||
)
|
||||
with patch.object(cache, "get", side_effect=redis_down), \
|
||||
patch.object(cache, "set", side_effect=redis_down):
|
||||
with self.assertLogs("core.models", level="WARNING") as logs:
|
||||
self.assertEqual(
|
||||
CoreSettings.get_default_user_agent_id(),
|
||||
"ua-from-db",
|
||||
)
|
||||
self.assertTrue(
|
||||
any("falling back to Postgres" in line for line in logs.output)
|
||||
)
|
||||
|
||||
def test_invalidate_tolerates_redis_unavailable(self):
|
||||
"""CoreSettings saves during migrate must not require Redis."""
|
||||
from redis.exceptions import ConnectionError as RedisConnectionError
|
||||
|
||||
redis_down = RedisConnectionError(
|
||||
"Error 111 connecting to localhost:6379"
|
||||
)
|
||||
with patch.object(cache, "get", side_effect=redis_down), \
|
||||
patch.object(cache, "set", side_effect=redis_down), \
|
||||
patch.object(cache, "delete", side_effect=redis_down):
|
||||
with self.assertLogs("core.models", level="WARNING") as logs:
|
||||
CoreSettings.invalidate_group_cache(SYSTEM_SETTINGS_KEY)
|
||||
self.assertTrue(
|
||||
any("falling back to Postgres" in line for line in logs.output)
|
||||
)
|
||||
|
||||
def test_cache_helpers_do_not_swallow_non_redis_errors(self):
|
||||
"""Programming errors and non-connectivity Redis errors still surface."""
|
||||
from redis.exceptions import AuthenticationError, ResponseError
|
||||
|
||||
with patch.object(cache, "get", side_effect=TypeError("boom")):
|
||||
with self.assertRaises(TypeError):
|
||||
CoreSettings._cache_get("any-key")
|
||||
with patch.object(cache, "set", side_effect=TypeError("boom")):
|
||||
with self.assertRaises(TypeError):
|
||||
CoreSettings._cache_set("any-key", {"a": 1})
|
||||
with patch.object(cache, "delete", side_effect=ValueError("boom")):
|
||||
with self.assertRaises(ValueError):
|
||||
CoreSettings._cache_delete("any-key")
|
||||
with patch.object(cache, "get", side_effect=ResponseError("WRONGTYPE")):
|
||||
with self.assertRaises(ResponseError):
|
||||
CoreSettings._cache_get("any-key")
|
||||
with patch.object(
|
||||
cache, "get", side_effect=AuthenticationError("NOAUTH")
|
||||
):
|
||||
with self.assertRaises(AuthenticationError):
|
||||
CoreSettings._cache_get("any-key")
|
||||
|
||||
def test_cache_backend_error_skips_fill(self):
|
||||
"""Failed ver/get must not collapse the version guard to None == None."""
|
||||
from redis.exceptions import ConnectionError as RedisConnectionError
|
||||
|
||||
CoreSettings.objects.create(
|
||||
key=SYSTEM_SETTINGS_KEY,
|
||||
name="System Settings",
|
||||
value={"catchup_enabled": True},
|
||||
)
|
||||
cache_key = CoreSettings.group_cache_key(SYSTEM_SETTINGS_KEY)
|
||||
cache.delete(cache_key)
|
||||
|
||||
def flaky_get(key, default=None, **kwargs):
|
||||
if key == cache_key:
|
||||
return None
|
||||
raise RedisConnectionError("flapping redis")
|
||||
|
||||
cached_sets = []
|
||||
real_set = cache.set
|
||||
|
||||
def tracking_set(key, value, timeout=None, **kwargs):
|
||||
cached_sets.append(key)
|
||||
return real_set(key, value, timeout=timeout, **kwargs)
|
||||
|
||||
with patch.object(cache, "get", side_effect=flaky_get), \
|
||||
patch.object(cache, "set", side_effect=tracking_set):
|
||||
self.assertTrue(CoreSettings.get_catchup_enabled())
|
||||
|
||||
self.assertNotIn(cache_key, cached_sets)
|
||||
|
||||
with patch.object(
|
||||
cache,
|
||||
"get",
|
||||
side_effect=RedisConnectionError("down"),
|
||||
):
|
||||
self.assertIs(
|
||||
CoreSettings._cache_get("any-key"),
|
||||
_CACHE_BACKEND_ERROR,
|
||||
)
|
||||
|
||||
|
||||
class DispatcharrUserAgentTests(TestCase):
|
||||
@patch('version.__version__', '1.2.3')
|
||||
|
|
|
|||
115
core/tests/test_migrate_without_redis.py
Normal file
115
core/tests/test_migrate_without_redis.py
Normal file
|
|
@ -0,0 +1,115 @@
|
|||
"""Fresh-install migrate must not require Redis (AIO boot order).
|
||||
|
||||
In AIO, ``manage.py migrate`` runs in the entrypoint before uWSGI starts
|
||||
Redis via ``attach-daemon``. Any data migration that hard-requires Redis
|
||||
leaves a partially migrated DB and a boot loop (see m3u.0003 / CoreSettings
|
||||
group cache).
|
||||
|
||||
This test creates a throwaway empty database and runs the full migration
|
||||
graph with the Django cache pointed at an unreachable Redis port. That is
|
||||
stronger than unit-testing a single settings helper: it catches any
|
||||
migration that starts depending on Redis.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from django.conf import settings
|
||||
from django.core.management import call_command
|
||||
from django.db import connection, connections
|
||||
from django.test import SimpleTestCase, override_settings
|
||||
|
||||
# Port 1 is never a Redis listener; Connection refused matches AIO first boot.
|
||||
_UNREACHABLE_REDIS_CACHES = {
|
||||
"default": {
|
||||
"BACKEND": "django_redis.cache.RedisCache",
|
||||
"LOCATION": "redis://127.0.0.1:1/0",
|
||||
"OPTIONS": {"CLIENT_CLASS": "django_redis.client.DefaultClient"},
|
||||
"TIMEOUT": 3600,
|
||||
}
|
||||
}
|
||||
|
||||
_TEMP_DB_NAME = "test_dispatcharr_migrate_noredis"
|
||||
|
||||
|
||||
def _admin_connect():
|
||||
import psycopg
|
||||
|
||||
db = settings.DATABASES["default"]
|
||||
return psycopg.connect(
|
||||
host=db["HOST"],
|
||||
port=db["PORT"],
|
||||
user=db["USER"],
|
||||
password=db["PASSWORD"],
|
||||
dbname="postgres",
|
||||
autocommit=True,
|
||||
)
|
||||
|
||||
|
||||
def _drop_database(name: str) -> None:
|
||||
admin = _admin_connect()
|
||||
try:
|
||||
admin.execute(
|
||||
"SELECT pg_terminate_backend(pid) FROM pg_stat_activity "
|
||||
"WHERE datname = %s AND pid <> pg_backend_pid()",
|
||||
(name,),
|
||||
)
|
||||
admin.execute(f'DROP DATABASE IF EXISTS "{name}"')
|
||||
finally:
|
||||
admin.close()
|
||||
|
||||
|
||||
def _create_empty_database(name: str) -> None:
|
||||
_drop_database(name)
|
||||
admin = _admin_connect()
|
||||
try:
|
||||
admin.execute(
|
||||
f'CREATE DATABASE "{name}" ENCODING \'UTF8\' TEMPLATE template0'
|
||||
)
|
||||
finally:
|
||||
admin.close()
|
||||
|
||||
|
||||
def _point_default_connection_at(name: str) -> None:
|
||||
"""Retarget the default DB alias without replacing settings.DATABASES.
|
||||
|
||||
``override_settings(DATABASES=...)`` leaves ConnectionHandler's cached
|
||||
settings pointing at the old dict, so migrate would silently keep using
|
||||
the already-migrated test database.
|
||||
"""
|
||||
connections.close_all()
|
||||
settings.DATABASES["default"]["NAME"] = name
|
||||
# connections.settings is the same dict after first configure; keep NAME
|
||||
# in sync if a copy was ever taken.
|
||||
connections.settings["default"]["NAME"] = name
|
||||
|
||||
|
||||
class FreshMigrateWithoutRedisTests(SimpleTestCase):
|
||||
"""Full migration suite against an empty DB while Redis is unreachable."""
|
||||
|
||||
databases = {"default"}
|
||||
|
||||
def test_full_migrate_succeeds_when_redis_is_unreachable(self):
|
||||
engine = settings.DATABASES["default"]["ENGINE"]
|
||||
if engine.endswith("sqlite3"):
|
||||
self.skipTest("Requires PostgreSQL (matches AIO fresh install)")
|
||||
|
||||
original_name = settings.DATABASES["default"]["NAME"]
|
||||
_create_empty_database(_TEMP_DB_NAME)
|
||||
|
||||
try:
|
||||
_point_default_connection_at(_TEMP_DB_NAME)
|
||||
with override_settings(CACHES=_UNREACHABLE_REDIS_CACHES):
|
||||
self.assertEqual(connection.settings_dict["NAME"], _TEMP_DB_NAME)
|
||||
# interactive=False: same as entrypoint ``migrate --noinput``.
|
||||
call_command("migrate", verbosity=0, interactive=False)
|
||||
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute("SELECT COUNT(*) FROM django_migrations")
|
||||
applied = cursor.fetchone()[0]
|
||||
# Sanity check: we actually migrated the empty DB, not a no-op
|
||||
# against the shared test database.
|
||||
self.assertGreater(applied, 50)
|
||||
finally:
|
||||
connections.close_all()
|
||||
_drop_database(_TEMP_DB_NAME)
|
||||
_point_default_connection_at(original_name)
|
||||
Loading…
Add table
Add a link
Reference in a new issue