@@ -101,11 +101,12 @@
hx-post="{% url 'run-cancel' run.id %}"
hx-headers='{"X-CSRFToken": "{{ csrf_token }}"}'
hx-swap="none"
+ hx-on::before-request="this.querySelector('button[type=submit]').disabled=true"
hx-on::after-request="if(event.detail.successful){ if(window._pwaNavigate) _pwaNavigate('{% url 'run-list' %}', true); else location.reload(); }"
@click.stop>
{% csrf_token %}
@@ -181,3 +182,24 @@
{% endblock %}
+
+{% block scripts %}
+
+{% endblock %}
diff --git a/webapp/templates/webapp/recipes/override_editor.html b/webapp/templates/webapp/recipes/override_editor.html
index 4534628..ead28a9 100644
--- a/webapp/templates/webapp/recipes/override_editor.html
+++ b/webapp/templates/webapp/recipes/override_editor.html
@@ -94,7 +94,7 @@
{{ t.RECIPES_VIE
+{% endblock %}
diff --git a/webapp/translations/en-US.json b/webapp/translations/en-US.json
index 2a5e6fe..608b4c6 100644
--- a/webapp/translations/en-US.json
+++ b/webapp/translations/en-US.json
@@ -9,6 +9,7 @@
"STATE_LIVE": "Live",
"STATE_PENDING": "Pending",
"STATE_CANCELLED": "Cancelled",
+ "STATE_CANCELLING": "Cancelling…",
"STATE_IN_PROGRESS": "In Progress",
"STATE_SKIPPED": "Skipped",
"STATE_DONE": "Done",
@@ -477,6 +478,8 @@
"FIND_LOADING": "Fetching recipe index…",
"FIND_EMPTY": "No recipes match your search.",
"FIND_EMPTY_QUERY": "Enter a search term above to find recipes.",
+ "FIND_INDEX_BUILDING": "Building recipe index…",
+ "FIND_INDEX_BUILDING_SUB": "This only happens once after the server starts.",
"FIND_ERROR": "Could not fetch the recipe index.",
"FIND_REFRESH_SUCCESS": "Recipe index refreshed successfully.",
"FIND_TOTAL": "results",
diff --git a/webapp/urls.py b/webapp/urls.py
index 39cdced..2264db9 100644
--- a/webapp/urls.py
+++ b/webapp/urls.py
@@ -27,8 +27,12 @@
runs.RunCancelView.as_view(), name='run-cancel'),
path('runs//',
runs.RunDetailView.as_view(), name='run-detail'),
+ path('runs//status/',
+ runs.run_status, name='run-status'),
path('runs//stream/',
runs.run_stream, name='run-stream'),
+ path('runs/stream/',
+ runs.run_list_stream, name='run-list-stream'),
path('config/schedule/',
schedule.ScheduleView.as_view(), name='schedule'),
path('config/about/',
diff --git a/webapp/views/runs.py b/webapp/views/runs.py
index 0eb43b2..1175f1e 100644
--- a/webapp/views/runs.py
+++ b/webapp/views/runs.py
@@ -1,6 +1,7 @@
import os
import plistlib
import ssl
+import threading
import time
import urllib.request
@@ -9,13 +10,28 @@
from django.contrib.auth.mixins import LoginRequiredMixin
from django.core.paginator import Paginator
from django.db import close_old_connections
-from django.http import StreamingHttpResponse, JsonResponse
+from django.http import StreamingHttpResponse, JsonResponse, HttpResponse
+
+
+class _AsyncStreamingHttpResponse(StreamingHttpResponse):
+ """Work around a Django 4.2 bug where StreamingHttpResponse.streaming_content
+ synchronously consumes async generators via async_to_sync even when is_async=True,
+ causing a spurious warning and defeating the async streaming. Overriding
+ __aiter__ to iterate _iterator directly bypasses the broken getter."""
+
+ async def __aiter__(self):
+ async for chunk in self._iterator:
+ yield self.make_bytes(chunk)
from django.shortcuts import get_object_or_404, redirect
from django.views import View
from django.views.generic import TemplateView
_MUNKI_CATALOG_CACHE: dict = {'data': None, 'url': '', 'auth': '', 'catalog': '', 'ts': 0.0}
+# Limit concurrent icon proxy requests to prevent exhausting the sync thread pool
+# that Django uses for non-async views under ASGI.
+_ICON_PROXY_SEMAPHORE = threading.Semaphore(3)
+
def _make_auth_header(username: str, password: str) -> str:
if not username:
@@ -212,72 +228,145 @@ def post(self, request, run_id):
run.status = 'cancelled'
run.completed_at = now
run.save(update_fields=['status', 'completed_at'])
+ # Signal the pipeline thread to stop and kill any running subprocess.
+ from webapp.runner import cancel_run
+ cancel_run(run_id)
if request.headers.get('HX-Request'):
from django.http import HttpResponse
- return HttpResponse(status=204)
+ resp = HttpResponse(status=204)
+ resp['HX-Trigger'] = 'runListRefresh'
+ return resp
return redirect('run-list')
@login_required
-@perm_required(PERM_VIEW_RUNS, PERM_TRIGGER_RUNS)
-def run_stream(request, run_id):
- """SSE endpoint: streams LogEntry rows for a running pipeline."""
- from webapp.models import LogEntry, Run, StageExecution
- import json
+def run_status(request, run_id):
+ """Lightweight JSON endpoint — returns run status and latest stage statuses.
- def event_stream():
- last_log_id = int(request.GET.get('from', 0))
- last_stage_data = {}
+ Used as an SSE fallback: when the SSE stream closes before the ``complete``
+ event is received, the JS polls this once to sync final state.
+ """
+ from webapp.models import Run, StageExecution
+ from webapp.perms import user_has_perm, PERM_VIEW_RUNS, PERM_TRIGGER_RUNS
+ if not (user_has_perm(request.user, PERM_VIEW_RUNS) or
+ user_has_perm(request.user, PERM_TRIGGER_RUNS)):
+ return JsonResponse({}, status=403)
+ run = Run.objects.filter(id=run_id).first()
+ if not run:
+ return JsonResponse({}, status=404)
+ stages = {
+ s.name: s.status
+ for s in StageExecution.objects.filter(run_id=run_id)
+ }
+ return JsonResponse({'status': run.status, 'stages': stages})
+
+
+async def run_stream(request, run_id):
+ """SSE endpoint — async, fan-out via RunBroadcaster.
+
+ Uses an async generator so each viewer is a cheap coroutine rather than
+ a blocked thread. All DB work is done by the broadcaster's single daemon
+ thread, so the database is polled once per second per run regardless of
+ how many clients are connected.
+ """
+ import asyncio
+ from asgiref.sync import sync_to_async
+ from webapp.run_broadcaster import broadcaster_manager
+ from webapp.perms import user_has_perm, PERM_VIEW_RUNS, PERM_TRIGGER_RUNS
+
+ # Inline auth — sync decorators don't compose cleanly with async views.
+ # auser() is only available on ASGIRequest; fall back under WSGI dev server.
+ # request.user is a SimpleLazyObject that hits the DB — must be evaluated
+ # in a thread via sync_to_async to satisfy Django's async safety guard.
+ if hasattr(request, 'auser'):
+ user = await request.auser()
+ else:
+ from django.contrib.auth import get_user as _get_user
+ user = await sync_to_async(_get_user)(request)
+ if not user.is_authenticated:
+ from django.contrib.auth.views import redirect_to_login
+ return redirect_to_login(request.get_full_path())
+ has_perm = await sync_to_async(
+ lambda: user_has_perm(user, PERM_VIEW_RUNS) or user_has_perm(user, PERM_TRIGGER_RUNS)
+ )()
+ if not has_perm:
+ return HttpResponse(status=403)
+
+ from webapp.models import Run
+ run_exists = await sync_to_async(Run.objects.filter(id=run_id).exists)()
+ if not run_exists:
+ return HttpResponse(status=404)
- while True:
- close_old_connections()
+ broadcaster = broadcaster_manager.get(run_id)
- run = Run.objects.filter(id=run_id).first()
- if not run:
- yield b'event: error\ndata: {"error": "run not found"}\n\n'
+ # The Last-Event-ID header is sent automatically by EventSource on reconnect.
+ try:
+ cursor = int(request.META.get('HTTP_LAST_EVENT_ID', -1))
+ except (ValueError, TypeError):
+ cursor = -1
+
+ async def event_stream():
+ nonlocal cursor
+ while True:
+ frames, done = broadcaster.events_since(cursor)
+ for frame in frames:
+ yield frame
+ cursor += 1
+ if done and not frames:
break
+ await asyncio.sleep(0.5)
- entries = LogEntry.objects.filter(
- run_id=run_id, id__gt=last_log_id
- ).order_by('id')
- for entry in entries:
- payload = json.dumps({
- 'type': 'log',
- 'level': entry.level,
- 'stage': entry.stage_name,
- 'message': entry.message,
- 'timestamp': entry.timestamp.isoformat(),
- })
- yield f'data: {payload}\n\n'.encode()
- last_log_id = entry.id
+ response = _AsyncStreamingHttpResponse(event_stream(), content_type='text/event-stream')
+ response['Cache-Control'] = 'no-cache'
+ response['X-Accel-Buffering'] = 'no'
+ # Prevent GZipMiddleware from buffering the stream before compressing it —
+ # SSE requires each event to be flushed immediately.
+ response['Content-Encoding'] = 'identity'
+ return response
- for stage in StageExecution.objects.filter(run_id=run_id):
- key = f'{stage.name}:{stage.status}'
- if last_stage_data.get(stage.name) != key:
- last_stage_data[stage.name] = key
- payload = json.dumps({
- 'type': 'stage',
- 'name': stage.name,
- 'status': stage.status,
- 'order': stage.order,
- 'started_at': stage.started_at.isoformat() if stage.started_at else None,
- 'completed_at': stage.completed_at.isoformat() if stage.completed_at else None,
- })
- yield f'data: {payload}\n\n'.encode()
-
- if run.status in ('success', 'failed', 'cancelled'):
- yield b'retry: 86400000\n\n'
- payload = json.dumps({'type': 'complete', 'status': run.status})
- yield f'data: {payload}\n\n'.encode()
- yield b'event: done\ndata: {}\n\n'
- break
- time.sleep(1)
+async def run_list_stream(request):
+ """SSE endpoint that fires a single event whenever any run reaches a terminal state.
- response = StreamingHttpResponse(event_stream(), content_type='text/event-stream')
+ The list page subscribes to this and uses the event to trigger an immediate
+ HTMX refresh of the run table, without waiting for the 10-second poll.
+ """
+ import asyncio
+ from asgiref.sync import sync_to_async
+ from webapp.run_broadcaster import run_list_broadcaster
+ from webapp.perms import user_has_perm, PERM_VIEW_RUNS, PERM_TRIGGER_RUNS
+
+ if hasattr(request, 'auser'):
+ user = await request.auser()
+ else:
+ from django.contrib.auth import get_user as _get_user
+ user = await sync_to_async(_get_user)(request)
+ if not user.is_authenticated:
+ from django.contrib.auth.views import redirect_to_login
+ return redirect_to_login(request.get_full_path())
+ has_perm = await sync_to_async(
+ lambda: user_has_perm(user, PERM_VIEW_RUNS) or user_has_perm(user, PERM_TRIGGER_RUNS)
+ )()
+ if not has_perm:
+ return HttpResponse(status=403)
+
+ async def event_stream():
+ last_seq = run_list_broadcaster._seq
+ while True:
+ new_seq = await asyncio.get_event_loop().run_in_executor(
+ None, run_list_broadcaster.wait_for_change, last_seq, 30.0
+ )
+ if new_seq > last_seq:
+ last_seq = new_seq
+ yield b'data: {}\n\n'
+ else:
+ yield b': keepalive\n\n'
+
+ response = _AsyncStreamingHttpResponse(event_stream(), content_type='text/event-stream')
response['Cache-Control'] = 'no-cache'
response['X-Accel-Buffering'] = 'no'
+ response['Content-Encoding'] = 'identity'
return response
@@ -300,11 +389,16 @@ def munki_icon_proxy(request):
Setting.get('repository.public_url_username', ''),
Setting.get('repository.public_url_password', ''),
)
+ if not _ICON_PROXY_SEMAPHORE.acquire(blocking=False):
+ return HttpResponse(status=503)
+
ctx = ssl.create_default_context(cafile=certifi.where())
headers = {'Authorization': auth_header} if auth_header else {}
- req = urllib.request.Request(f'{public_url}/{path}', headers=headers)
+ from urllib.parse import quote
+ encoded_path = quote(path, safe='/')
+ req = urllib.request.Request(f'{public_url}/{encoded_path}', headers=headers)
try:
- with urllib.request.urlopen(req, timeout=10, context=ctx) as r:
+ with urllib.request.urlopen(req, timeout=4, context=ctx) as r:
content_type = r.headers.get('Content-Type', 'image/png')
data = r.read()
response = HttpResponse(data, content_type=content_type)
@@ -312,3 +406,5 @@ def munki_icon_proxy(request):
return response
except Exception:
return HttpResponse(status=404)
+ finally:
+ _ICON_PROXY_SEMAPHORE.release()