mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-18 00:55:50 +00:00
fix(schema): resolve Swagger/OpenAPI schema generation issues under concurrent requests
- Implemented a new `LockedSpectacularAPIView` to handle schema generation, ensuring it no longer fails due to concurrent gevent requests. The schema builds are now single-flight per process and cached in Django's cache, allowing all workers to share the result. - Updated the `urls.py` to use the new view for schema generation, enhancing stability and performance.
This commit is contained in:
SergeantPanda
parent
388c4bd28d
commit
cdfe16ae5d
5 changed files with 298 additions and 19 deletions
|
|
@ -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).
|
||||
|
|
|
|||
173
apps/api/schema_views.py
Normal file
173
apps/api/schema_views.py
Normal file
|
|
@ -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)}"'
|
||||
)
|
||||
},
|
||||
)
|
||||
|
|
@ -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'),
|
||||
]
|
||||
|
|
|
|||
|
|
@ -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: {
|
||||
|
|
|
|||
103
tests/test_openapi_schema.py
Normal file
103
tests/test_openapi_schema.py
Normal file
|
|
@ -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))
|
||||
Loading…
Add table
Add a link
Reference in a new issue