From e17f4f64a98278ff4d7c060c5ec5bb90a46c8713 Mon Sep 17 00:00:00 2001 From: busya Date: Sun, 31 May 2026 17:17:30 +0300 Subject: [PATCH] websocket add --- backend/src/api/routes/maintenance/_routes.py | 23 ++ backend/src/app.py | 215 +++++++++++++++--- backend/src/core/task_manager/event_bus.py | 103 ++++++++- backend/src/core/task_manager/lifecycle.py | 30 +++ backend/src/core/task_manager/manager.py | 46 ++++ frontend/src/components/TaskHistory.svelte | 98 ++++++-- frontend/src/components/TaskLogViewer.svelte | 18 +- .../__tests__/task_log_viewer.test.js | 2 +- frontend/src/lib/api.js | 32 +++ .../lib/components/layout/TaskDrawer.svelte | 57 +++-- .../components/reports/reportTypeProfiles.js | 2 - frontend/src/lib/stores/maintenance.svelte.js | 99 +++++--- frontend/src/routes/maintenance/+page.svelte | 2 +- frontend/tests/maintenance.test.ts | 4 +- 14 files changed, 590 insertions(+), 141 deletions(-) diff --git a/backend/src/api/routes/maintenance/_routes.py b/backend/src/api/routes/maintenance/_routes.py index a5a28b4f..84d8808e 100644 --- a/backend/src/api/routes/maintenance/_routes.py +++ b/backend/src/api/routes/maintenance/_routes.py @@ -350,6 +350,17 @@ async def start_maintenance( extra={"event_id": event.id, "task_id": task.id}, ) + # ── Broadcast maintenance event ── + task_manager.broadcast_maintenance_event({ + "type": "maintenance.event_created", + "maintenance_id": event.id, + "tables": sorted_tables, + "start_time": event.start_time.isoformat() if event.start_time else None, + "end_time": event.end_time.isoformat() if event.end_time else None, + "message": message, + "environment_id": env_id, + }) + return MaintenanceStartResponse( task_id=task.id, maintenance_id=event.id, @@ -421,6 +432,12 @@ async def end_maintenance( extra={"event_id": maintenance_id, "task_id": task.id}, ) + # ── Broadcast maintenance event ── + task_manager.broadcast_maintenance_event({ + "type": "maintenance.event_ended", + "maintenance_id": maintenance_id, + }) + return MaintenanceEndResponse( task_id=task.id, status="pending", @@ -485,6 +502,12 @@ async def end_all_maintenance( extra={"task_id": task.id, "environment_id": effective_env_id}, ) + # ── Broadcast maintenance event ── + task_manager.broadcast_maintenance_event({ + "type": "maintenance.events_ended_all", + "environment_id": effective_env_id or "", + }) + return MaintenanceEndResponse( task_id=task.id, status="pending", diff --git a/backend/src/app.py b/backend/src/app.py index 509de40d..41f6d147 100755 --- a/backend/src/app.py +++ b/backend/src/app.py @@ -470,7 +470,10 @@ async def websocket_endpoint( websocket: WebSocket, task_id: str, source: str = None, level: str = None ): """ - WebSocket endpoint for real-time log streaming with optional server-side filtering. + WebSocket endpoint for real-time log streaming AND task status updates. + Sends two message types: + - Log entries: plain dicts with level/message/timestamp (backward compatible, no type field) + - Status updates: dict with type="task_status" and nested task dict Query Parameters: source: Filter logs by source component (e.g., "plugin", "superset_api") level: Filter logs by minimum level (DEBUG, INFO, WARNING, ERROR) @@ -490,7 +493,7 @@ async def websocket_endpoint( level_hierarchy = {"DEBUG": 0, "INFO": 1, "WARNING": 2, "ERROR": 3} min_level = level_hierarchy.get(level_filter, 0) if level_filter else 0 logger.reason( - "Accepted WebSocket log stream connection", + "Accepted WebSocket log+status stream connection", extra={ "task_id": task_id, "source_filter": source_filter, @@ -499,11 +502,13 @@ async def websocket_endpoint( }, ) task_manager = get_task_manager() - queue = await task_manager.subscribe_logs(task_id) + log_queue = await task_manager.subscribe_logs(task_id) + status_queue = await task_manager.subscribe_status(task_id) logger.reason( - "Subscribed WebSocket client to task log queue", + "Subscribed WebSocket client to task log and status queues", extra={"task_id": task_id}, ) + def matches_filters(log_entry) -> bool: """Check if log entry matches the filter criteria.""" log_source = getattr(log_entry, "source", None) @@ -514,7 +519,37 @@ async def websocket_endpoint( if log_level < min_level: return False return True + + async def send_status_update(task: Any) -> None: + """Send a structured task status update over the WebSocket.""" + status_dict = { + "id": task.id, + "plugin_id": task.plugin_id, + "status": task.status.value if hasattr(task.status, "value") else str(task.status), + "started_at": task.started_at.isoformat() if task.started_at else None, + "finished_at": task.finished_at.isoformat() if task.finished_at else None, + "user_id": task.user_id, + "result": task.result, + "input_required": task.input_required, + "input_request": task.input_request, + } + await websocket.send_json({ + "type": "task_status", + "task_id": task.id, + "task": status_dict, + }) + try: + # ── Send initial task status ── + task = task_manager.get_task(task_id) + if task: + await send_status_update(task) + logger.reason( + "Sent initial task status", + extra={"task_id": task_id, "status": str(task.status)}, + ) + + # ── Replay initial logs ── logger.reason( "Starting task log stream replay and live forwarding", extra={"task_id": task_id}, @@ -535,6 +570,8 @@ async def websocket_endpoint( "total_available_logs": len(initial_logs), }, ) + + # ── Send synthetic AWAITING_INPUT prompt if needed ── task = task_manager.get_task(task_id) if task and task.status == "AWAITING_INPUT" and task.input_request: synthetic_log = { @@ -550,47 +587,167 @@ async def websocket_endpoint( "Replayed awaiting-input prompt to restored WebSocket client", extra={"task_id": task_id, "task_status": task.status}, ) + + # ── Main loop: listen on both log and status queues ── while True: - log_entry = await queue.get() - if not matches_filters(log_entry): - continue - log_dict = log_entry.model_dump() - log_dict["timestamp"] = log_dict["timestamp"].isoformat() - await websocket.send_json(log_dict) - logger.reflect( - "Forwarded task log entry to WebSocket client", - extra={ - "task_id": task_id, - "level": log_dict.get("level"), - }, + done, _ = await asyncio.wait( + [log_queue.get(), status_queue.get()], + return_when=asyncio.FIRST_COMPLETED, ) - if ( - "Task completed successfully" in log_entry.message - or "Task failed" in log_entry.message - ): - logger.reason( - "Observed terminal task log entry; delaying to preserve client visibility", - extra={"task_id": task_id, "message": log_entry.message}, + for coro in done: + result = coro.result() + + # ── Status update ── + if isinstance(result, dict) and result.get("type") == "task_status": + await websocket.send_json(result) + task_status = result.get("task", {}).get("status", "") + if task_status in ("SUCCESS", "FAILED"): + logger.reason( + "Task reached terminal state via status broadcast; closing stream", + extra={"task_id": task_id, "status": task_status}, + ) + await asyncio.sleep(2) + raise StopIteration # exit the while loop + continue + + # ── Log entry ── + if not matches_filters(result): + continue + log_dict = result.model_dump() + log_dict["timestamp"] = log_dict["timestamp"].isoformat() + await websocket.send_json(log_dict) + logger.reflect( + "Forwarded task log entry to WebSocket client", + extra={ + "task_id": task_id, + "level": log_dict.get("level"), + }, ) - await asyncio.sleep(2) - except WebSocketDisconnect: + if ( + "Task completed successfully" in result.message + or "Task failed" in result.message + ): + logger.reason( + "Observed terminal task log entry; delaying to preserve client visibility", + extra={"task_id": task_id, "message": result.message}, + ) + await asyncio.sleep(2) + except (WebSocketDisconnect, StopIteration): + if isinstance(StopIteration): + pass # normal termination logger.reason( - "WebSocket client disconnected from task log stream", + "WebSocket client disconnected or stream ended", extra={"task_id": task_id}, ) except Exception as exc: logger.explore( - "WebSocket log streaming encountered an unexpected failure", + "WebSocket log+status streaming encountered an unexpected failure", extra={"task_id": task_id, "error": str(exc)}, ) raise finally: - task_manager.unsubscribe_logs(task_id, queue) + task_manager.unsubscribe_logs(task_id, log_queue) + task_manager.unsubscribe_status(task_id, status_queue) logger.reflect( - "Released WebSocket log queue subscription", + "Released WebSocket log and status queue subscriptions", extra={"task_id": task_id}, ) # #endregion websocket_endpoint + +# #region task_events_websocket [C:4] [TYPE Function] +# @BRIEF WebSocket endpoint for global task events (status changes for ALL tasks). +# @RELATION CALLS -> [TaskManagerPackage] +# @PRE WebSocket must be authenticated via `token` query param. +# @POST WebSocket streams task status events until disconnect. +@app.websocket("/ws/task-events") +async def task_events_websocket(websocket: WebSocket): + """ + WebSocket endpoint for global task events. + Streams {type: "task_status", task_id: ..., task: {...}} for ALL task status changes. + Query Parameters: + token: JWT or API key for authentication (required) + """ + seed_trace_id() + with belief_scope("task_events_websocket"): + if not await _authenticate_websocket(websocket, "ws/task-events"): + await websocket.close(code=4001, reason="Authentication required") + return + + await websocket.accept() + logger.reason("Accepted global task events WebSocket connection") + + task_manager = get_task_manager() + event_queue = await task_manager.subscribe_task_events() + logger.reason("Subscribed to global task events") + + try: + while True: + event = await event_queue.get() + await websocket.send_json(event) + logger.reflect( + "Forwarded task event to global client", + extra={"task_id": event.get("task_id")}, + ) + except WebSocketDisconnect: + logger.reason("Global task events client disconnected") + except Exception as exc: + logger.explore( + "Global task events streaming failed", + extra={"error": str(exc)}, + ) + raise + finally: + task_manager.unsubscribe_task_events(event_queue) + logger.reflect("Released global task events subscription") +# #endregion task_events_websocket + +# #region maintenance_events_websocket [C:4] [TYPE Function] +# @BRIEF WebSocket endpoint for maintenance events (created/ended/banner changes). +# @RELATION CALLS -> [TaskManagerPackage] +# @PRE WebSocket must be authenticated via `token` query param. +# @POST WebSocket streams maintenance events until disconnect. +@app.websocket("/ws/maintenance/events") +async def maintenance_events_websocket(websocket: WebSocket): + """ + WebSocket endpoint for maintenance events. + Streams {type: "maintenance.event_created", ...} / {type: "maintenance.event_ended"}. + Query Parameters: + token: JWT or API key for authentication (required) + """ + seed_trace_id() + with belief_scope("maintenance_events_websocket"): + if not await _authenticate_websocket(websocket, "ws/maintenance/events"): + await websocket.close(code=4001, reason="Authentication required") + return + + await websocket.accept() + logger.reason("Accepted maintenance events WebSocket connection") + + task_manager = get_task_manager() + event_queue = await task_manager.subscribe_maintenance_events() + logger.reason("Subscribed to maintenance events") + + try: + while True: + event = await event_queue.get() + await websocket.send_json(event) + logger.reflect( + "Forwarded maintenance event to client", + extra={"event_type": event.get("type"), "maintenance_id": event.get("maintenance_id")}, + ) + except WebSocketDisconnect: + logger.reason("Maintenance events client disconnected") + except Exception as exc: + logger.explore( + "Maintenance events streaming failed", + extra={"error": str(exc)}, + ) + raise + finally: + task_manager.unsubscribe_maintenance_events(event_queue) + logger.reflect("Released maintenance events subscription") +# #endregion maintenance_events_websocket + # #region dataset_websocket_endpoint [C:4] [TYPE Function] # @BRIEF WebSocket endpoint for dataset.updated events — auto-refresh on task completion. # @RELATION CALLS -> [TaskManagerPackage] diff --git a/backend/src/core/task_manager/event_bus.py b/backend/src/core/task_manager/event_bus.py index be085c8e..86962c14 100644 --- a/backend/src/core/task_manager/event_bus.py +++ b/backend/src/core/task_manager/event_bus.py @@ -49,11 +49,17 @@ class EventBus: LOG_FLUSH_INTERVAL = 2.0 # #region __init__ [TYPE Function] - # @BRIEF Initialize empty log buffer, subscriber registry, and start background flusher. + # @BRIEF Initialize empty log buffer, subscriber registry, status subscribers, and start background flusher. def __init__(self, log_persistence_service: TaskLogPersistenceService, loop: asyncio.AbstractEventLoop): self.log_persistence_service = log_persistence_service self.loop = loop self.subscribers: dict[str, list[asyncio.Queue]] = {} + # Per-task status subscribers (same pattern as log subscribers) + self._status_subscribers: dict[str, list[asyncio.Queue]] = {} + # Global task event subscribers (all tasks) + self._task_event_subscribers: list[asyncio.Queue] = [] + # Maintenance event subscribers + self._maintenance_subscribers: list[asyncio.Queue] = [] self._log_buffer: dict[str, list[LogEntry]] = {} self._log_buffer_lock = threading.Lock() self._flusher_stop_event = threading.Event() @@ -178,6 +184,101 @@ class EventBus: del self.subscribers[task_id] # #endregion unsubscribe_logs + # ── Task Status Subscribers ── + + # #region subscribe_status [C:2] [TYPE Function] [SEMANTICS subscribe,status,realtime] + # @BRIEF Subscribes to real-time status updates for a specific task. + # @POST Returns an asyncio.Queue for Task status dicts. + async def subscribe_status(self, task_id: str) -> asyncio.Queue: + queue = asyncio.Queue() + if task_id not in self._status_subscribers: + self._status_subscribers[task_id] = [] + self._status_subscribers[task_id].append(queue) + return queue + # #endregion subscribe_status + + # #region unsubscribe_status [C:2] [TYPE Function] [SEMANTICS unsubscribe,status,remove] + # @BRIEF Unsubscribes from real-time status updates for a task. + # @POST Queue removed from status subscribers. + def unsubscribe_status(self, task_id: str, queue: asyncio.Queue): + if task_id in self._status_subscribers: + if queue in self._status_subscribers[task_id]: + self._status_subscribers[task_id].remove(queue) + if not self._status_subscribers[task_id]: + del self._status_subscribers[task_id] + # #endregion unsubscribe_status + + # #region broadcast_status [C:2] [TYPE Function] [SEMANTICS broadcast,status,notify] + # @BRIEF Broadcast a task status update to all per-task and global subscribers. + # @POST Status event pushed to all matching queues. + def broadcast_status(self, task_id: str, task_dict: dict): + from ..logger import logger as _lg + event = {"type": "task_status", "task_id": task_id, "task": task_dict} + # Per-task subscribers + if task_id in self._status_subscribers: + for queue in self._status_subscribers[task_id]: + try: + self.loop.call_soon_threadsafe(queue.put_nowait, event) + except Exception as _e: + _lg.warning("Failed to send status to per-task subscriber", extra={"task_id": task_id, "error": str(_e)}) + # Global task event subscribers + for queue in self._task_event_subscribers: + try: + self.loop.call_soon_threadsafe(queue.put_nowait, event) + except Exception as _e: + _lg.warning("Failed to send status to global subscriber", extra={"task_id": task_id, "error": str(_e)}) + # #endregion broadcast_status + + # ── Maintenance Event Subscribers ── + + # #region subscribe_maintenance_events [C:2] [TYPE Function] [SEMANTICS subscribe,maintenance,events] + # @BRIEF Subscribes to maintenance event updates (created/ended/banner changes). + # @POST Returns an asyncio.Queue for maintenance event dicts. + async def subscribe_maintenance_events(self) -> asyncio.Queue: + queue = asyncio.Queue() + self._maintenance_subscribers.append(queue) + return queue + # #endregion subscribe_maintenance_events + + # #region unsubscribe_maintenance_events [C:2] [TYPE Function] [SEMANTICS unsubscribe,maintenance,events] + # @BRIEF Unsubscribes from maintenance events. + # @POST Queue removed from maintenance subscribers. + def unsubscribe_maintenance_events(self, queue: asyncio.Queue): + if queue in self._maintenance_subscribers: + self._maintenance_subscribers.remove(queue) + # #endregion unsubscribe_maintenance_events + + # #region broadcast_maintenance_event [C:2] [TYPE Function] [SEMANTICS broadcast,maintenance,notify] + # @BRIEF Broadcast a maintenance event to all subscribers. + # @POST Event pushed to all maintenance subscriber queues. + def broadcast_maintenance_event(self, event: dict): + from ..logger import logger as _lg + for queue in self._maintenance_subscribers: + try: + self.loop.call_soon_threadsafe(queue.put_nowait, event) + except Exception as _e: + _lg.warning("Failed to send maintenance event to subscriber", extra={"event_type": event.get("type"), "error": str(_e)}) + # #endregion broadcast_maintenance_event + + # ── Global Task Event Subscribers ── + + # #region subscribe_task_events [C:2] [TYPE Function] [SEMANTICS subscribe,global,events] + # @BRIEF Subscribes to ALL task events (status changes) globally. + # @POST Returns an asyncio.Queue for task event dicts. + async def subscribe_task_events(self) -> asyncio.Queue: + queue = asyncio.Queue() + self._task_event_subscribers.append(queue) + return queue + # #endregion subscribe_task_events + + # #region unsubscribe_task_events [C:2] [TYPE Function] [SEMANTICS unsubscribe,global,events] + # @BRIEF Unsubscribes from global task events. + # @POST Queue removed from global subscribers. + def unsubscribe_task_events(self, queue: asyncio.Queue): + if queue in self._task_event_subscribers: + self._task_event_subscribers.remove(queue) + # #endregion unsubscribe_task_events + # #region get_task_logs [C:3] [TYPE Function] [SEMANTICS logs,read,persistence,backfill] # @BRIEF Retrieves logs for a specific task (from memory for running, persistence for completed). # @RELATION CALLS -> [EXT:method:TaskLogPersistenceService.get_logs] diff --git a/backend/src/core/task_manager/lifecycle.py b/backend/src/core/task_manager/lifecycle.py index 7d1f6e89..636feab5 100644 --- a/backend/src/core/task_manager/lifecycle.py +++ b/backend/src/core/task_manager/lifecycle.py @@ -34,6 +34,22 @@ from .context import TaskContext from .graph import TaskGraph from .models import Task, TaskStatus +def _task_to_dict(task: Task) -> dict: + """Serialize a Task to a dict for WebSocket broadcast.""" + return { + "id": task.id, + "plugin_id": task.plugin_id, + "status": task.status.value if isinstance(task.status, TaskStatus) else task.status, + "started_at": task.started_at.isoformat() if task.started_at else None, + "finished_at": task.finished_at.isoformat() if task.finished_at else None, + "user_id": task.user_id, + "params": task.params, + "result": task.result, + "input_required": task.input_required, + "input_request": task.input_request, + "error": str(task.result) if task.status == TaskStatus.FAILED and task.result else None, + } + # #region JobLifecycle [C:5] [TYPE Class] [SEMANTICS task,lifecycle,execution,state,machine] # @BRIEF Encodes task creation, execution, pause/resume, and completion transitions for @@ -129,6 +145,7 @@ class JobLifecycle: task.status = TaskStatus.RUNNING task.started_at = datetime.now(UTC) self.persistence_service.persist_task(task) + self._broadcast_task_status(task) if add_log_callback: add_log_callback( task_id, "INFO", @@ -164,6 +181,7 @@ class JobLifecycle: ) logger.info(f"Task {task_id} completed successfully") task.status = TaskStatus.SUCCESS + self._broadcast_task_status(task) if add_log_callback: add_log_callback( task_id, "INFO", @@ -173,6 +191,7 @@ class JobLifecycle: except Exception as e: logger.error(f"Task {task_id} failed: {e}") task.status = TaskStatus.FAILED + self._broadcast_task_status(task) if add_log_callback: add_log_callback( task_id, "ERROR", @@ -214,6 +233,7 @@ class JobLifecycle: task.params.update(resolution_params) task.status = TaskStatus.RUNNING self.persistence_service.persist_task(task) + self._broadcast_task_status(task) self.graph.resolve_future(task_id, True) # #endregion resolve_task @@ -228,6 +248,7 @@ class JobLifecycle: return task.status = TaskStatus.AWAITING_MAPPING self.persistence_service.persist_task(task) + self._broadcast_task_status(task) future = self.loop.create_future() self.graph.create_future(task_id, future) try: @@ -274,6 +295,7 @@ class JobLifecycle: task.input_required = True task.input_request = input_request self.persistence_service.persist_task(task) + self._broadcast_task_status(task) if add_log_callback: add_log_callback( task_id, "INFO", @@ -308,6 +330,7 @@ class JobLifecycle: task.input_request = None task.status = TaskStatus.RUNNING self.persistence_service.persist_task(task) + self._broadcast_task_status(task) if add_log_callback: add_log_callback( task_id, "INFO", @@ -337,6 +360,13 @@ class JobLifecycle: del self._dataset_subscribers[env_id] # #endregion unsubscribe_dataset_events + # #region _broadcast_task_status [C:2] [TYPE Function] [SEMANTICS broadcast,status,websocket] + # @BRIEF Broadcast task status update to all subscribers via EventBus. + # @POST EventBus.broadcast_status called with serialized task dict. + def _broadcast_task_status(self, task: Task): + self.event_bus.broadcast_status(task.id, _task_to_dict(task)) + # #endregion _broadcast_task_status + # #region _broadcast_dataset_updated [C:3] [TYPE Function] [SEMANTICS dataset,broadcast,event,websocket] # @BRIEF Broadcast dataset.updated event to all subscribers for a given environment. def _broadcast_dataset_updated(self, env_id: str, dataset_ids: list[int]): diff --git a/backend/src/core/task_manager/manager.py b/backend/src/core/task_manager/manager.py index 17c219a5..2912b73f 100644 --- a/backend/src/core/task_manager/manager.py +++ b/backend/src/core/task_manager/manager.py @@ -261,6 +261,32 @@ class TaskManager: self.event_bus.unsubscribe_logs(task_id, queue) # #endregion unsubscribe_logs + # ── Status subscribers ── + + # #region subscribe_status [TYPE Function] [C:2] + # @BRIEF Subscribes to real-time status updates for a task. + async def subscribe_status(self, task_id: str) -> asyncio.Queue: + return await self.event_bus.subscribe_status(task_id) + # #endregion subscribe_status + + # #region unsubscribe_status [TYPE Function] [C:2] + # @BRIEF Unsubscribes from status updates for a task. + def unsubscribe_status(self, task_id: str, queue: asyncio.Queue): + self.event_bus.unsubscribe_status(task_id, queue) + # #endregion unsubscribe_status + + # #region subscribe_task_events [TYPE Function] [C:2] + # @BRIEF Subscribes to global task events (all task status changes). + async def subscribe_task_events(self) -> asyncio.Queue: + return await self.event_bus.subscribe_task_events() + # #endregion subscribe_task_events + + # #region unsubscribe_task_events [TYPE Function] [C:2] + # @BRIEF Unsubscribes from global task events. + def unsubscribe_task_events(self, queue: asyncio.Queue): + self.event_bus.unsubscribe_task_events(queue) + # #endregion unsubscribe_task_events + # ── Lifecycle delegates to JobLifecycle ── # #region create_task [TYPE Function] [C:4] @@ -318,6 +344,26 @@ class TaskManager: ) # #endregion resume_task_with_password + # ── Maintenance event delegates to EventBus ── + + # #region subscribe_maintenance_events [TYPE Function] [C:2] + # @BRIEF Subscribes to global maintenance events. + async def subscribe_maintenance_events(self) -> asyncio.Queue: + return await self.event_bus.subscribe_maintenance_events() + # #endregion subscribe_maintenance_events + + # #region unsubscribe_maintenance_events [TYPE Function] [C:2] + # @BRIEF Unsubscribes from global maintenance events. + def unsubscribe_maintenance_events(self, queue: asyncio.Queue): + self.event_bus.unsubscribe_maintenance_events(queue) + # #endregion unsubscribe_maintenance_events + + # #region broadcast_maintenance_event [TYPE Function] [C:2] + # @BRIEF Broadcast a maintenance event to all subscribers. + def broadcast_maintenance_event(self, event: dict): + self.event_bus.broadcast_maintenance_event(event) + # #endregion broadcast_maintenance_event + # ── Dataset event delegates to JobLifecycle ── # #region subscribe_dataset_events [TYPE Function] [C:2] diff --git a/frontend/src/components/TaskHistory.svelte b/frontend/src/components/TaskHistory.svelte index c08f5ea4..2fa5684b 100644 --- a/frontend/src/components/TaskHistory.svelte +++ b/frontend/src/components/TaskHistory.svelte @@ -11,7 +11,7 @@ diff --git a/frontend/src/components/TaskLogViewer.svelte b/frontend/src/components/TaskLogViewer.svelte index 8da63835..8c07fed3 100644 --- a/frontend/src/components/TaskLogViewer.svelte +++ b/frontend/src/components/TaskLogViewer.svelte @@ -12,7 +12,6 @@ @INVARIANT: Real-time logs are always appended without duplicates. --> {#if shouldShow} diff --git a/frontend/src/components/__tests__/task_log_viewer.test.js b/frontend/src/components/__tests__/task_log_viewer.test.js index bc86ec17..6aef74b4 100644 --- a/frontend/src/components/__tests__/task_log_viewer.test.js +++ b/frontend/src/components/__tests__/task_log_viewer.test.js @@ -3,7 +3,7 @@ // @PURPOSE: Unit tests for TaskLogViewer component by mounting it and observing the DOM. // @LAYER UI (Tests) // @RELATION DEPENDS_ON -> [TaskLogViewer] -// @INVARIANT: Duplicate logs are never appended. Polling only active for in-progress tasks. +// @INVARIANT: Duplicate logs are never appended. Historical logs loaded once on open; real-time logs via prop. // @TEST_CONTRACT: TaskLogViewerPropsAndLogStream -> RenderedLogTimeline // @TEST_SCENARIO: historical_and_realtime_merge -> Historical logs render and realtime logs append without duplication. // @TEST_FIXTURE: valid_viewer -> INLINE_JSON diff --git a/frontend/src/lib/api.js b/frontend/src/lib/api.js index 55fda126..f466559c 100755 --- a/frontend/src/lib/api.js +++ b/frontend/src/lib/api.js @@ -59,6 +59,38 @@ export const getWsUrl = (taskId) => { return url; }; +/** + * Build a WebSocket URL for global task events, including auth token. + */ +export const getTaskEventsWsUrl = () => { + const protocol = typeof window !== 'undefined' && window.location.protocol === 'https:' ? 'wss:' : 'ws:'; + const host = typeof window !== 'development' && typeof window !== 'undefined' ? window.location.host : 'localhost:8000'; + let url = `${protocol}//${host}/ws/task-events`; + if (typeof window !== 'undefined') { + const token = localStorage.getItem('auth_token'); + if (token) { + url += `?token=${encodeURIComponent(token)}`; + } + } + return url; +}; + +/** + * Build a WebSocket URL for maintenance events, including auth token. + */ +export const getMaintenanceEventsWsUrl = () => { + const protocol = typeof window !== 'undefined' && window.location.protocol === 'https:' ? 'wss:' : 'ws:'; + const host = typeof window !== 'development' && typeof window !== 'undefined' ? window.location.host : 'localhost:8000'; + let url = `${protocol}//${host}/ws/maintenance/events`; + if (typeof window !== 'undefined') { + const token = localStorage.getItem('auth_token'); + if (token) { + url += `?token=${encodeURIComponent(token)}`; + } + } + return url; +}; + /** * Build a WebSocket URL for translation run progress streaming. * The token is appended as a query param for WebSocket auth. diff --git a/frontend/src/lib/components/layout/TaskDrawer.svelte b/frontend/src/lib/components/layout/TaskDrawer.svelte index b6e4e291..5fcb7a11 100644 --- a/frontend/src/lib/components/layout/TaskDrawer.svelte +++ b/frontend/src/lib/components/layout/TaskDrawer.svelte @@ -66,13 +66,13 @@ import { gitService } from "../../../services/gitService.js"; let ws = null; + let wsReconnectTimer = null; let realTimeLogs = $state([]); let taskStatus = $state(null); let recentTasks = $state([]); let loadingTasks = $state(false); let activeTaskDetails = $state(null); let environmentOptions = []; - let taskDetailsPollInterval = null; let diffText = $state(""); let showDiff = $state(false); let isDiffLoading = $state(false); @@ -168,13 +168,6 @@ return "bg-slate-100 text-slate-700 ring-1 ring-slate-200"; } - function stopTaskDetailsPolling() { - if (taskDetailsPollInterval) { - clearInterval(taskDetailsPollInterval); - taskDetailsPollInterval = null; - } - } - async function loadEnvironmentOptions() { try { const response = await api.getEnvironments(); @@ -222,21 +215,6 @@ return `${baseUrl}/superset/dashboard/${encodeURIComponent(String(dashboardId))}/`; } - async function loadActiveTaskDetails() { - const taskId = normalizeTaskId(activeTaskId); - if (!taskId) return; - try { - const task = await api.getTask(taskId); - activeTaskDetails = task; - taskStatus = task?.status || taskStatus; - if (isTaskFinished(task?.status)) { - stopTaskDetailsPolling(); - } - } catch (err) { - console.error("[TaskDrawer][loadActiveTaskDetails][FAILED]", err); - } - } - function extractPrimaryDashboardId(task) { const result = task?.result || {}; const params = task?.params || {}; @@ -426,8 +404,18 @@ ws.onmessage = (event) => { const data = JSON.parse(event.data); - console.log(`[TaskDrawer][WebSocket][Message_Received] ${data.message}`); + // ── Task status update from WebSocket ── + if (data.type === "task_status") { + const taskData = data.task; + console.log(`[TaskDrawer][WebSocket][Status_Update] ${taskData.status}`, taskData); + activeTaskDetails = taskData; + taskStatus = taskData.status; + return; + } + + // ── Log entry (backward compatible, no type field) ── + console.log(`[TaskDrawer][WebSocket][Log_Received] ${data.message}`); realTimeLogs = [...realTimeLogs, data]; if (data.message?.includes("Task completed successfully")) { @@ -442,7 +430,11 @@ }; ws.onclose = () => { - console.log("[TaskDrawer][WebSocket] Connection closed"); + console.log("[TaskDrawer][WebSocket] Disconnected — reconnecting in 10s"); + ws = null; + wsReconnectTimer = setTimeout(() => { + if (activeTaskId) connectWebSocket(); + }, 10000); }; } @@ -456,7 +448,12 @@ */ function disconnectWebSocket() { console.log("[TaskDrawer][WebSocket][disconnectWebSocket:START]"); + if (wsReconnectTimer) { + clearTimeout(wsReconnectTimer); + wsReconnectTimer = null; + } if (ws) { + ws.onclose = null; // prevent reconnect ws.close(); ws = null; } @@ -532,18 +529,14 @@ showDiff = false; diffText = ""; taskStatus = "RUNNING"; + activeTaskDetails = null; connectWebSocket(); void loadEnvironmentOptions(); - void loadActiveTaskDetails(); - stopTaskDetailsPolling(); - taskDetailsPollInterval = setInterval(loadActiveTaskDetails, 4000); } else { - stopTaskDetailsPolling(); activeTaskDetails = null; void loadRecentTasks(); } } else { - stopTaskDetailsPolling(); activeTaskDetails = null; } }); @@ -555,7 +548,9 @@ // Cleanup on destroy onDestroy(() => { - stopTaskDetailsPolling(); + if (wsReconnectTimer) { + clearTimeout(wsReconnectTimer); + } disconnectWebSocket(); }); diff --git a/frontend/src/lib/components/reports/reportTypeProfiles.js b/frontend/src/lib/components/reports/reportTypeProfiles.js index 77b27038..6ca6a2c9 100644 --- a/frontend/src/lib/components/reports/reportTypeProfiles.js +++ b/frontend/src/lib/components/reports/reportTypeProfiles.js @@ -59,8 +59,6 @@ export const REPORT_TYPE_PROFILES = { // @TEST_INVARIANT: fallbacks_to_unknown -> verifies: [invalid_type] export function getReportTypeProfile(taskType) { const key = typeof taskType === 'string' ? taskType : 'unknown'; - console.log("[reports][ui][getReportTypeProfile][STATE:START]"); - console.log("[reports][ui][getReportTypeProfile] Resolved type '" + taskType + "' to profile '" + key + "'"); return REPORT_TYPE_PROFILES[key] || REPORT_TYPE_PROFILES.unknown; } // #endregion getReportTypeProfile:Function diff --git a/frontend/src/lib/stores/maintenance.svelte.js b/frontend/src/lib/stores/maintenance.svelte.js index db076ac9..300dafa3 100644 --- a/frontend/src/lib/stores/maintenance.svelte.js +++ b/frontend/src/lib/stores/maintenance.svelte.js @@ -1,20 +1,21 @@ -// #region MaintenanceStore [C:3] [TYPE Module] [SEMANTICS store, maintenance, svelte, runes] -// @BRIEF Svelte 5 (Runes) store for maintenance banner state. Auto-polls events every 30s via interval. +// #region MaintenanceStore [C:3] [TYPE Module] [SEMANTICS store, maintenance, svelte, runes, websocket] +// @BRIEF Svelte 5 (Runes) store for maintenance banner state. Uses WebSocket for real-time events. // @LAYER Frontend // @RELATION DEPENDS_ON -> [MaintenanceApi] // @RELATION DEPENDS_ON -> [EXT:frontend:addToast] +// @RELATION DEPENDS_ON -> [EXT:frontend:getMaintenanceEventsWsUrl] // @UI_STATE Idle -> events loaded, settings loaded, banners loaded // @UI_STATE Loading -> isLoading = true during fetches // @UI_STATE Error -> error state populated with message -// @ADR-0007: Uses $effect(() => subscribe(...)) pattern, NOT fromStore + $derived. -// @UX_REACTIVITY $state for events/settings/banners. Manual setInterval for polling. -// @UX_RECOVERY Error toast on failure. Retry via loadEvents/loadSettings. +// @UX_REACTIVITY $state for events/settings/banners. WebSocket for real-time updates. +// @UX_RECOVERY WebSocket reconnect after 10s on disconnect. Fallback to HTTP init(). +// @RATIONALE Replaced 30s polling with WebSocket for instant maintenance banner updates +// on the dashboard hub page and maintenance events page. import { addToast } from "$lib/toasts.js"; +import { getMaintenanceEventsWsUrl } from "$lib/api.js"; import * as maintenanceApi from "$lib/api/maintenance.js"; -const POLL_INTERVAL_MS = 30000; - /** * @returns {object} Maintenance store with runes state and methods. */ @@ -26,11 +27,64 @@ export function createMaintenanceStore() { let dashboardBanners = $state([]); let isLoading = $state(false); let error = $state(null); - let _pollInterval = null; + let _ws = null; // #endregion MaintenanceStore.state // #region MaintenanceStore.methods [C:2] [TYPE Block] + /** + * Connect to WebSocket for real-time maintenance events. + * Replaces the 30s polling interval. + */ + function connectWs() { + if (_ws) return; // already connected + try { + const url = getMaintenanceEventsWsUrl(); + _ws = new WebSocket(url); + + _ws.onopen = () => { + console.log("[MaintenanceStore][WS] Connected"); + }; + + _ws.onmessage = async (event) => { + try { + const data = JSON.parse(event.data); + console.log("[MaintenanceStore][WS] Event:", data.type); + // On any maintenance event, reload events and dashboard banners + await Promise.all([loadEvents(), loadDashboardBanners()]); + } catch (_e) { + // Ignore parse errors + } + }; + + _ws.onerror = () => { + console.warn("[MaintenanceStore][WS] Error"); + }; + + _ws.onclose = () => { + console.log("[MaintenanceStore][WS] Disconnected — reconnecting in 10s"); + _ws = null; + setTimeout(() => { + if (_ws) return; // already reconnected + connectWs(); + }, 10000); + }; + } catch (_e) { + _ws = null; + } + } + + /** + * Disconnect WebSocket. + */ + function disconnectWs() { + if (_ws) { + _ws.onclose = null; // prevent reconnect + _ws.close(); + _ws = null; + } + } + /** * Load events from API. */ @@ -84,6 +138,7 @@ export function createMaintenanceStore() { if (data?.task_id) { addToast("Removal task started", "info"); } + // Immediate optimistic refresh; WS will also trigger reload await loadEvents(); } catch (e) { const msg = e?.message || "Failed to end maintenance event"; @@ -101,6 +156,7 @@ export function createMaintenanceStore() { if (data?.task_id) { addToast("Bulk removal task started", "info"); } + // Immediate optimistic refresh; WS will also trigger reload await loadEvents(); } catch (e) { const msg = e?.message || "Failed to end all events"; @@ -128,31 +184,11 @@ export function createMaintenanceStore() { } /** - * Initialize the store: load all data once and start polling. + * Initialize the store: load all data once and connect WebSocket. */ async function init() { await Promise.all([loadEvents(), loadSettings(), loadDashboardBanners()]); - startPolling(); - } - - /** - * Start auto-polling for events every 30s. - */ - function startPolling() { - if (_pollInterval) return; - _pollInterval = setInterval(async () => { - await loadEvents(); - }, POLL_INTERVAL_MS); - } - - /** - * Stop auto-polling. - */ - function stopPolling() { - if (_pollInterval) { - clearInterval(_pollInterval); - _pollInterval = null; - } + connectWs(); } // #endregion MaintenanceStore.methods @@ -171,8 +207,7 @@ export function createMaintenanceStore() { endAllEvents, updateSettings, init, - startPolling, - stopPolling, + disconnectWs, }; } diff --git a/frontend/src/routes/maintenance/+page.svelte b/frontend/src/routes/maintenance/+page.svelte index ef7d6345..13348cfc 100644 --- a/frontend/src/routes/maintenance/+page.svelte +++ b/frontend/src/routes/maintenance/+page.svelte @@ -30,7 +30,7 @@ }); onDestroy(() => { - maintenanceStore.stopPolling(); + maintenanceStore.disconnectWs(); }); diff --git a/frontend/tests/maintenance.test.ts b/frontend/tests/maintenance.test.ts index 19f3b5a4..33fa4a1d 100644 --- a/frontend/tests/maintenance.test.ts +++ b/frontend/tests/maintenance.test.ts @@ -30,8 +30,8 @@ const { mockStoreState, mockStore } = vi.hoisted(() => { endAllEvents: vi.fn(), updateSettings: vi.fn(), init: vi.fn(), - startPolling: vi.fn(), - stopPolling: vi.fn(), + connectWs: vi.fn(), + disconnectWs: vi.fn(), }; return { mockStoreState: sharedState, mockStore: store };