diff --git a/CHANGELOG.md b/CHANGELOG.md index 86fe5205..1173ce29 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,6 +43,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **Swagger/OpenAPI schema generation no longer fails under concurrent gevent requests.** Concurrent hits to `/api/schema/` (for example Swagger UI loading) could race on DRF's shared `AutoSchema` state and raise `AssertionError: Schema generation REQUIRES a view instance`, leaving Swagger empty. Schema builds are now single-flight per process and the result is cached in Django's cache (Redis) so all workers share it. - **`/proxy/ts/change_stream` now persists `stream_id` and waits for the real switch outcome in multi-worker deployments.** When the request landed on a worker that didn't own the channel, the owning worker's `STREAM_SWITCH` pubsub handler performed the switch but wrote only `url`/`user_agent` back to the channel's Redis metadata, so `GET /proxy/ts/status` kept reporting the old `stream_id` indefinitely; the handler now persists the full switch metadata (`stream_id`, `m3u_profile`, stream name) via the same `_update_channel_metadata()` helper the direct owner path uses. `stream_name` from the initial `get_stream_info_for_switch()` lookup is now forwarded through the pubsub payload (and from `change_stream` / `next_stream` on the direct owner path) so the owner no longer re-queries the database when writing metadata. The endpoint also no longer returns an optimistic 200 on the non-owner path: the requesting worker waits (up to 15s) for the owner to confirm via the `switch_status` Redis key and answers 502 if the owner reports failure or 504 if no confirmation arrives (e.g. owner worker gone); `next_stream` gets the same treatment. Requesting a switch to the URL the channel is already playing is now treated as success (with a metadata refresh) instead of a silent no-op, and the `switch_status` key now expires (60s TTL) instead of persisting forever. (Fixes #1412) - **XC live streams, `/panel_api.php`, and `/xmltv.php` no longer crash when a visible channel has no channel number.** Since `channel_number` became nullable, auto-sync and compact numbering can leave a channel with `effective_channel_number=None` while it remains visible in output (`hidden_from_output=False`). The XC live-stream number map skipped those rows but still tried to emit them, causing `KeyError` on `get_live_streams` and a 500 on `/panel_api.php` (`xc_get_info`). Authenticated XMLTV export (`/xmltv.php`, profile EPG with a logged-in user) had the same gap and could fail chunk-cache rebuilds with `KeyError` on `channel_num_map[channel.id]`. Null-number channels are now deferred into the existing collision-free integer assignment pass (same second loop as fractional numbers) and receive the next available integer; `_xc_channel_entry` also falls back to `channel.id` if a row is still unmapped. HDHR lineup behaviour is unchanged (null-number channels remain omitted there). - **System events and Connect dispatch no longer close the DB connection mid-call outside worker contexts.** `log_system_event()` and `dispatch_event_system()` always called `close_old_connections()` in `finally` blocks, which broke Django `TestCase` transactions and could interrupt in-flight ORM work when Connect/plugin handlers ran synchronously. Closes are now limited to gevent uWSGI workers (and the Celery sync-dispatch path where gevent is patched but no hub runs). Celery workers still reset connections via existing `task_postrun` / `task_prerun` hooks on the standard PostgreSQL backend (no geventpool). diff --git a/apps/api/schema_views.py b/apps/api/schema_views.py new file mode 100644 index 00000000..85297193 --- /dev/null +++ b/apps/api/schema_views.py @@ -0,0 +1,173 @@ +"""OpenAPI schema views that are safe under gevent concurrency. + +DRF's AutoSchema descriptor keeps mutable per-class state (``self.view``). +When uWSGI serves requests with gevent, concurrent ``/api/schema/`` builds +interleave on that state and raise: + + AssertionError: Schema generation REQUIRES a view instance + +Only one schema build runs at a time per process (single-flight). The +finished schema is stored in Django's cache (Redis) so all workers share it. +Waiters block on an Event instead of holding a lock across DB I/O, which +avoids gevent deadlocks with the connection pool. +""" + +from __future__ import annotations + +import copy +import logging +import threading +from typing import Dict, Optional + +from django.conf import settings +from django.core.cache import cache +from rest_framework.response import Response + +from drf_spectacular.views import SpectacularAPIView + +logger = logging.getLogger(__name__) + +# Greenlet-safe after gevent monkey-patching. +_guard = threading.Lock() +_in_progress: Dict[str, threading.Event] = {} +# Same-process handoff when cache.set fails so waiters are not left empty-handed. +_flight_results: Dict[str, dict] = {} + +_SCHEMA_CACHE_PREFIX = "openapi:schema:" +_SCHEMA_CACHE_VER_KEY = f"{_SCHEMA_CACHE_PREFIX}cache_ver" + + +def _cache_supports_delete_pattern() -> bool: + return callable(getattr(cache, "delete_pattern", None)) + + +def clear_schema_cache() -> None: + """Invalidate cached schemas (tests / forced refresh after deploy).""" + with _guard: + _flight_results.clear() + try: + if _cache_supports_delete_pattern(): + cache.delete_pattern(f"{_SCHEMA_CACHE_PREFIX}*") + return + ver = cache.get(_SCHEMA_CACHE_VER_KEY) or 0 + cache.set(_SCHEMA_CACHE_VER_KEY, int(ver) + 1, timeout=None) + except Exception: + logger.warning("Failed to clear OpenAPI schema cache", exc_info=True) + + +def _key_namespace() -> str: + """Stable on Redis (pattern delete); versioned on LocMem for test clears.""" + if _cache_supports_delete_pattern(): + return "shared" + try: + ver = cache.get(_SCHEMA_CACHE_VER_KEY) + if ver is None: + cache.add(_SCHEMA_CACHE_VER_KEY, 1, timeout=None) + ver = cache.get(_SCHEMA_CACHE_VER_KEY) or 1 + return f"v{int(ver)}" + except Exception: + return "v1" + + +class LockedSpectacularAPIView(SpectacularAPIView): + """SpectacularAPIView with single-flight generation and Django cache.""" + + def _resolve_version(self, request): + return self.api_version or request.version or self._get_version_parameter( + request + ) + + def _cache_key(self, request) -> str: + version = self._resolve_version(request) + lang = request.GET.get("lang") if settings.USE_I18N else None + urlconf = self.urlconf + if urlconf is None: + urlconf_part = "default" + else: + urlconf_part = getattr(urlconf, "__name__", None) or type(urlconf).__name__ + patterns_part = "custom" if self.patterns is not None else "default" + return ( + f"{_SCHEMA_CACHE_PREFIX}{_key_namespace()}:" + f"{version}:{lang}:{self.serve_public}:{urlconf_part}:{patterns_part}" + ) + + def _cache_get(self, key: str) -> Optional[dict]: + try: + return cache.get(key) + except Exception: + logger.warning("OpenAPI schema cache get failed", exc_info=True) + return None + + def _cache_set(self, key: str, schema: dict) -> None: + try: + # Use CACHES['default'] TIMEOUT (3600s in settings). + cache.set(key, schema) + except Exception: + logger.warning("OpenAPI schema cache set failed", exc_info=True) + + def _get_schema_response(self, request): + version = self._resolve_version(request) + cache_key = self._cache_key(request) + + cached = self._cache_get(cache_key) + if cached is not None: + return self._schema_response(request, version, cached) + + with _guard: + cached = self._cache_get(cache_key) + if cached is not None: + return self._schema_response(request, version, cached) + + event = _in_progress.get(cache_key) + if event is None: + event = threading.Event() + _in_progress[cache_key] = event + is_builder = True + else: + is_builder = False + + if not is_builder: + if not event.wait(timeout=120): + return Response( + {"detail": "Timed out waiting for OpenAPI schema generation."}, + status=503, + ) + cached = self._cache_get(cache_key) + if cached is None: + with _guard: + cached = _flight_results.get(cache_key) + if cached is None: + return Response( + {"detail": "OpenAPI schema generation failed."}, + status=500, + ) + return self._schema_response(request, version, cached) + + try: + generator = self.generator_class( + urlconf=self.urlconf, + api_version=version, + patterns=self.patterns, + ) + schema = generator.get_schema( + request=request, public=self.serve_public + ) + with _guard: + _flight_results[cache_key] = schema + self._cache_set(cache_key, schema) + return self._schema_response(request, version, schema) + finally: + with _guard: + _in_progress.pop(cache_key, None) + event.set() + + def _schema_response(self, request, version, schema: dict) -> Response: + # LocMem returns the same object; deepcopy keeps responses isolated. + return Response( + data=copy.deepcopy(schema), + headers={ + "Content-Disposition": ( + f'inline; filename="{self._get_filename(request, version)}"' + ) + }, + ) diff --git a/apps/api/urls.py b/apps/api/urls.py index 5af829d3..200dc0a6 100644 --- a/apps/api/urls.py +++ b/apps/api/urls.py @@ -1,5 +1,7 @@ from django.urls import path, include, re_path -from drf_spectacular.views import SpectacularAPIView, SpectacularSwaggerView, SpectacularRedocView +from drf_spectacular.views import SpectacularSwaggerView, SpectacularRedocView + +from apps.api.schema_views import LockedSpectacularAPIView app_name = 'api' @@ -22,9 +24,9 @@ urlpatterns = [ - # OpenAPI Schema and Documentation (drf-spectacular) - path('schema/', SpectacularAPIView.as_view(), name='schema'), + # OpenAPI schema (single-flight + Django cache; see apps.api.schema_views) + path('schema/', LockedSpectacularAPIView.as_view(), name='schema'), re_path(r'^swagger/?$', SpectacularSwaggerView.as_view(url_name='api:schema'), name='swagger-ui'), path('redoc/', SpectacularRedocView.as_view(url_name='api:schema'), name='redoc'), - path('swagger.json', SpectacularAPIView.as_view(), name='schema-json'), + path('swagger.json', LockedSpectacularAPIView.as_view(), name='schema-json'), ] diff --git a/frontend/vite.config.js b/frontend/vite.config.js index 1026e519..3e8996fa 100644 --- a/frontend/vite.config.js +++ b/frontend/vite.config.js @@ -10,21 +10,21 @@ export default defineConfig({ server: { port: 9191, - - // proxy: { - // "/api": { - // target: "http://localhost:5656", // Backend server - // changeOrigin: true, - // secure: false, // Set to true if backend uses HTTPS - // // rewrite: (path) => path.replace(/^\/api/, ""), // Optional path rewrite - // }, - // "/ws": { - // target: "http://localhost:8001", // Backend server - // changeOrigin: true, - // secure: false, // Set to true if backend uses HTTPS - // // rewrite: (path) => path.replace(/^\/api/, ""), // Optional path rewrite - // }, - // }, + // Without this, /api/* is served as the React SPA in debug mode and + // Swagger UI at /api/swagger/ never loads the OpenAPI schema. + proxy: { + "/api": { + target: "http://127.0.0.1:5656", + changeOrigin: true, + secure: false, + }, + "/ws": { + target: "http://127.0.0.1:8001", + changeOrigin: true, + secure: false, + ws: true, + }, + }, }, test: { diff --git a/tests/test_openapi_schema.py b/tests/test_openapi_schema.py new file mode 100644 index 00000000..9281c2f5 --- /dev/null +++ b/tests/test_openapi_schema.py @@ -0,0 +1,103 @@ +from concurrent.futures import ThreadPoolExecutor, as_completed +import threading +import time +from unittest.mock import patch + +from django.test import SimpleTestCase, RequestFactory, override_settings + +from apps.api.schema_views import LockedSpectacularAPIView, clear_schema_cache + + +def _counting_generator(calls): + class CountingGenerator: + def __init__(self, *args, **kwargs): + pass + + def get_schema(self, request=None, public=True): + calls["n"] += 1 + return {"openapi": "3.0.3", "paths": {}, "info": {"title": "t"}} + + return CountingGenerator + + +@override_settings( + CACHES={ + "default": { + "BACKEND": "django.core.cache.backends.locmem.LocMemCache", + "LOCATION": "openapi-schema-tests", + } + } +) +class LockedSpectacularAPIViewTests(SimpleTestCase): + def setUp(self): + clear_schema_cache() + self.factory = RequestFactory() + + def tearDown(self): + clear_schema_cache() + + def test_schema_cache_is_reused(self): + view = LockedSpectacularAPIView.as_view() + calls = {"n": 0} + + with patch.object( + LockedSpectacularAPIView, "generator_class", _counting_generator(calls) + ): + for _ in range(3): + response = view(self.factory.get("/api/schema/")) + self.assertEqual(response.status_code, 200) + + self.assertEqual(calls["n"], 1) + + def test_clear_schema_cache_forces_rebuild(self): + view = LockedSpectacularAPIView.as_view() + calls = {"n": 0} + + with patch.object( + LockedSpectacularAPIView, "generator_class", _counting_generator(calls) + ): + self.assertEqual(view(self.factory.get("/api/schema/")).status_code, 200) + clear_schema_cache() + self.assertEqual(view(self.factory.get("/api/schema/")).status_code, 200) + + self.assertEqual(calls["n"], 2) + + def test_concurrent_first_load_uses_single_build(self): + view = LockedSpectacularAPIView.as_view() + call_count = 0 + in_flight = 0 + max_in_flight = 0 + counter_lock = threading.Lock() + started = threading.Barrier(8) + + class SlowGenerator: + def __init__(self, *args, **kwargs): + pass + + def get_schema(self, request=None, public=True): + nonlocal call_count, in_flight, max_in_flight + with counter_lock: + call_count += 1 + in_flight += 1 + max_in_flight = max(max_in_flight, in_flight) + time.sleep(0.05) + with counter_lock: + in_flight -= 1 + return {"openapi": "3.0.3", "paths": {}, "info": {"title": "t"}} + + with patch.object( + LockedSpectacularAPIView, "generator_class", SlowGenerator + ): + def hit(_): + started.wait(timeout=5) + response = view(self.factory.get("/api/schema/")) + self.assertEqual(response.status_code, 200) + return response.data + + with ThreadPoolExecutor(max_workers=8) as pool: + futures = [pool.submit(hit, i) for i in range(8)] + results = [f.result(timeout=10) for f in as_completed(futures)] + + self.assertEqual(call_count, 1) + self.assertEqual(max_in_flight, 1) + self.assertTrue(all(r.get("openapi") == "3.0.3" for r in results))