From d3d51780dd74848b072f89c6e5ea51bcc9dccdcb Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 16 Jul 2026 19:24:46 +0000 Subject: [PATCH] fix(schema): enhance Swagger/OpenAPI schema generation for concurrent requests --- CHANGELOG.md | 2 +- apps/api/schema_views.py | 55 ++++++++++++++++++++++++++++++---------- 2 files changed, 42 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1173ce29..adf6df28 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,7 +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. +- **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. Cold builds could also hang the gevent uWSGI worker. Schema introspection now runs in a real OS thread, builds are 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 index 85297193..3aef9ec4 100644 --- a/apps/api/schema_views.py +++ b/apps/api/schema_views.py @@ -6,10 +6,10 @@ 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. +Cold builds also frequently hang the gevent hub when introspection runs in a +greenlet. Generation therefore runs in a real OS thread (gevent ThreadPool), +with per-process single-flight coordination and a Django/Redis cache shared +by all workers. """ from __future__ import annotations @@ -17,7 +17,7 @@ from __future__ import annotations import copy import logging import threading -from typing import Dict, Optional +from typing import Callable, Dict, Optional from django.conf import settings from django.core.cache import cache @@ -27,14 +27,32 @@ from drf_spectacular.views import SpectacularAPIView logger = logging.getLogger(__name__) +try: + # Real OS threads; .get() waits without blocking the gevent hub. + from gevent import monkey + from gevent.threadpool import ThreadPool + + _SCHEMA_POOL: Optional[ThreadPool] = ThreadPool(maxsize=1) + _GEVENT_PATCHED = monkey.is_module_patched("threading") +except ImportError: # pragma: no cover - non-gevent runtimes + _SCHEMA_POOL = None + _GEVENT_PATCHED = False + # 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" +_BUILD_TIMEOUT_SECONDS = 120 + + +def _run_schema_build(build: Callable[[], dict]) -> dict: + """Run schema introspection off the gevent hub when monkey-patched.""" + if _SCHEMA_POOL is not None and _GEVENT_PATCHED: + return _SCHEMA_POOL.spawn(build).get(timeout=_BUILD_TIMEOUT_SECONDS) + return build() def _cache_supports_delete_pattern() -> bool: @@ -105,6 +123,14 @@ class LockedSpectacularAPIView(SpectacularAPIView): except Exception: logger.warning("OpenAPI schema cache set failed", exc_info=True) + def _build_schema(self, request, version: Optional[str]) -> dict: + generator = self.generator_class( + urlconf=self.urlconf, + api_version=version, + patterns=self.patterns, + ) + return generator.get_schema(request=request, public=self.serve_public) + def _get_schema_response(self, request): version = self._resolve_version(request) cache_key = self._cache_key(request) @@ -127,7 +153,7 @@ class LockedSpectacularAPIView(SpectacularAPIView): is_builder = False if not is_builder: - if not event.wait(timeout=120): + if not event.wait(timeout=_BUILD_TIMEOUT_SECONDS): return Response( {"detail": "Timed out waiting for OpenAPI schema generation."}, status=503, @@ -144,18 +170,19 @@ class LockedSpectacularAPIView(SpectacularAPIView): 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 + schema = _run_schema_build( + lambda: self._build_schema(request, version) ) with _guard: _flight_results[cache_key] = schema self._cache_set(cache_key, schema) return self._schema_response(request, version, schema) + except Exception: + logger.exception("OpenAPI schema generation failed") + return Response( + {"detail": "OpenAPI schema generation failed."}, + status=500, + ) finally: with _guard: _in_progress.pop(cache_key, None)