marker: route marker.match to a dedicated project WS channel

High-frequency marker.matches shared the single project notification queue with topology events (node.*/link.*), causing head-of-line blocking. Add a separate marker channel: Notification.project_marker_queue/marker_emit, dispatch routes marker.* off the main project queue, plus a new WS /{project_id}/notifications/markers/ws endpoint. Fully migrated (the main project WS no longer carries marker.match); marker listeners are independent of project auto_close. Compute side unchanged.
This commit is contained in:
YueGuobin committed 2026-08-11 11:19:56 +08:00
1 parent a699383933
commit 2ac0bb7d25
3 files changed
+101

No files matched your search

@@ -485,6 +485,38 @@ async def project_ws_notifications(
await project.close()
@router.websocket("/{project_id}/notifications/markers/ws")
async def project_marker_ws_notifications(
project_id: UUID,
websocket: WebSocket,
current_user: schemas.User = Depends(has_privilege_on_websocket("Project.Audit"))
) -> None:
"""
Receive marker notifications (e.g. marker.match) for a project on a
dedicated WebSocket, separate from the main project stream so high-frequency
marker.matches do not block topology events (node.*/link.*).
Required privilege: Project.Audit
"""
if current_user is None:
return
controller = Controller.instance()
project = controller.get_project(str(project_id))
log.info(f"New client has connected to the marker notification stream for project ID '{project.id}' (WebSocket method)")
try:
with controller.notification.project_marker_queue(project.id) as queue:
while True:
notification = await queue.get_json(5)
await websocket.send_text(notification)
except (ConnectionClosed, WebSocketDisconnect):
log.info(f"Client has disconnected from the marker notification stream for project ID '{project.id}' (WebSocket method)")
except WebSocketException as e:
log.warning(f"Error while sending marker event to WebSocket client: {e}")
@router.get("/{project_id}/export", dependencies=[Depends(has_privilege("Project.Audit"))])
async def export_project(
project: Project = Depends(dep_project),
+42
View File
@@ -31,6 +31,7 @@ class Notification:
self._controller = controller
self._project_listeners = {}
self._project_marker_listeners = {}
self._controller_listeners = set()
@contextmanager
@@ -49,6 +50,26 @@ class Notification:
finally:
self._project_listeners[project_id].remove(queue)
@contextmanager
def project_marker_queue(self, project_id):
"""
Get a queue of marker notifications (marker.match etc.) for a project.
Marker events are delivered on this dedicated channel instead of the
main project queue, so high-frequency marker.matches do not cause
head-of-line blocking for topology events (node.*/link.*).
Use it with Python with
"""
queue = NotificationQueue()
self._project_marker_listeners.setdefault(project_id, set())
self._project_marker_listeners[project_id].add(queue)
try:
yield queue
finally:
self._project_marker_listeners[project_id].remove(queue)
@contextmanager
def controller_queue(self):
"""
@@ -104,6 +125,8 @@ class Notification:
elif action == "ping":
event["compute_id"] = compute_id
self.project_emit(action, event)
elif action.startswith("marker."):
self.marker_emit(action, event, project_id)
else:
self.project_emit(action, event, project_id)
@@ -120,6 +143,25 @@ class Notification:
else:
self._send_event_to_all_projects(action, event)
def marker_emit(self, action, event, project_id):
"""
Send a marker notification (e.g. marker.match) to clients listening on
the dedicated marker channel for this project. Marker events are kept
off the main project queue on purpose, to avoid head-of-line blocking
from high-frequency matches.
:param action: Action name
:param event: Event to send
:param project_id: Project id the marker belongs to
"""
try:
marker_listeners = self._project_marker_listeners[project_id]
except KeyError:
return
for listener in marker_listeners:
asyncio.get_running_loop().call_soon_threadsafe(listener.put_nowait, (action, event, {}))
def _send_event_to_project(self, project_id, action, event):
"""
Send an event to all the client listening for notifications for