diff --git a/CHANGELOG.md b/CHANGELOG.md index 7cf8327e..f799c2f9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/core/models.py b/core/models.py index 17be7c96..ef14d39e 100644 --- a/core/models.py +++ b/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 diff --git a/core/tests/test_core.py b/core/tests/test_core.py index c5a236d0..5c337e24 100644 --- a/core/tests/test_core.py +++ b/core/tests/test_core.py @@ -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') diff --git a/core/tests/test_migrate_without_redis.py b/core/tests/test_migrate_without_redis.py new file mode 100644 index 00000000..9312666c --- /dev/null +++ b/core/tests/test_migrate_without_redis.py @@ -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)