Push SSE history updates to non-owner viewers

Owner dispatch only delivers history_update events to the user who owns
the changed history. Watching a shared or pinned non-owned history
(e.g. via the multi-history view) silently received nothing under SSE
because the dispatch path keys on history ownership and there was no
way for a viewer to opt in.

Add an additive viewer-subscription channel: the client POSTs the
history ids it wants pushes for to /api/events/history-subscriptions,
the dispatcher fans the record out to every webapp worker over Kombu,
and history_update fanout pushes to subscribed viewers in addition to
the owner. Subscriptions are refcounted on the client and replayed on
every EventSource onopen so reconnects don't drop them. Per-worker
subscription state is purged when a user's last connection on that
worker closes; the client's onopen replay re-asserts the desired set.

No authorization check on the dispatch path: SSE events only carry
history ids, the client follows up with a normal access-controlled
REST fetch, and leaking "history changed at T" pings is acceptable.

MultipleViewItem now subscribes whenever it renders a history the
current user does not own and SSE is enabled; owned histories continue
through owner dispatch with zero subscribe round-trip.

Tests: integration test exercising subscribe -> mutate -> receive ->
unsubscribe -> stop receiving end-to-end, plus vitest for the client
refcount and replay logic.

Author: mvdbeek

client/src/components/History/Multiple/MultipleViewItem.vue

@@ -3,10 +3,17 @@ import { faTimes } from "@fortawesome/free-solid-svg-icons";
 import { FontAwesomeIcon } from "@fortawesome/vue-fontawesome";
 import { BButton } from "bootstrap-vue";
 import { storeToRefs } from "pinia";
-import { computed, ref } from "vue";
+import { computed, ref, watch } from "vue";
 
+import { userOwnsHistory } from "@/api";
 import { useExtendedHistory } from "@/composables/detailedHistory";
+import {
+    addHistoryViewerSubscription,
+    removeHistoryViewerSubscription,
+} from "@/composables/useNotificationSSE";
+import { useConfigStore } from "@/stores/configurationStore";
 import { useHistoryStore } from "@/stores/historyStore";
+import { useUserStore } from "@/stores/userStore";
 
 import CollectionPanel from "@/components/History/CurrentCollection/CollectionPanel.vue";
 import HistoryNavigation from "@/components/History/CurrentHistory/HistoryNavigation.vue";
@@ -24,6 +31,8 @@ const props = defineProps<Props>();
 
 const historyStore = useHistoryStore();
 const { currentHistoryId, pinnedHistories } = storeToRefs(historyStore);
+const { currentUser } = storeToRefs(useUserStore());
+const configStore = useConfigStore();
 
 const { history } = useExtendedHistory(props.source.id);
 
@@ -33,6 +42,31 @@ const sameToCurrent = computed(() => {
     return currentHistoryId.value === props.source.id;
 });
 
+// Owner routing covers histories the current user owns; only non-owned
+// histories need a viewer subscription. Skip in polling mode — there's no
+// SSE channel to push events to and the REST call would still hit the queue.
+const needsViewerSubscription = computed(() => {
+    if (!configStore.config?.enable_sse_updates) {
+        return false;
+    }
+    if (!history.value || !currentUser.value) {
+        return false;
+    }
+    return !userOwnsHistory(currentUser.value, history.value);
+});
+
+watch(
+    [needsViewerSubscription, () => props.source.id],
+    ([subscribe, id], _previous, onCleanup) => {
+        if (!subscribe || !id) {
+            return;
+        }
+        addHistoryViewerSubscription(id);
+        onCleanup(() => removeHistoryViewerSubscription(id));
+    },
+    { immediate: true },
+);
+
 function onViewCollection(collection: object) {
     selectedCollections.value = [...selectedCollections.value, collection];
 }

client/src/composables/useNotificationSSE.test.ts

@@ -0,0 +1,99 @@
+/**
+ * Unit tests for the viewer-subscription helpers in useNotificationSSE.
+ *
+ * The shared EventSource isn't exercised here — these tests focus on the
+ * client-side bookkeeping: refcounting, dedup, and HTTP shape. Reconnect
+ * replay is covered by an MSW request log assertion plus an explicit call
+ * into the (test-only) onopen replay path.
+ */
+
+import { http, HttpResponse } from "msw";
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+
+import { useServerMock } from "@/api/client/__mocks__";
+
+import {
+    _resetHistoryViewerSubscriptionsForTest,
+    addHistoryViewerSubscription,
+    removeHistoryViewerSubscription,
+} from "./useNotificationSSE";
+
+interface SubscriptionRequest {
+    method: "POST" | "DELETE";
+    history_ids: string[];
+}
+
+const { server } = useServerMock();
+
+describe("useNotificationSSE viewer subscriptions", () => {
+    let requests: SubscriptionRequest[];
+
+    beforeEach(() => {
+        _resetHistoryViewerSubscriptionsForTest();
+        requests = [];
+        const recordHandler = async ({ request }: { request: Request }) => {
+            const body = (await request.json()) as { history_ids: string[] };
+            requests.push({
+                method: request.method as "POST" | "DELETE",
+                history_ids: body.history_ids,
+            });
+            return new HttpResponse(null, { status: 204 });
+        };
+        server.use(
+            http.post("/api/events/history-subscriptions", recordHandler),
+            http.delete("/api/events/history-subscriptions", recordHandler),
+        );
+    });
+
+    afterEach(() => {
+        _resetHistoryViewerSubscriptionsForTest();
+    });
+
+    it("POSTs once per first subscriber for a given history id", async () => {
+        addHistoryViewerSubscription("hist-A");
+        await Promise.resolve();
+        expect(requests).toHaveLength(1);
+        expect(requests[0]?.method).toBe("POST");
+        expect(requests[0]?.history_ids).toEqual(["hist-A"]);
+    });
+
+    it("refcounts duplicate subscriptions — second add is a no-op on the wire", async () => {
+        addHistoryViewerSubscription("hist-A");
+        addHistoryViewerSubscription("hist-A");
+        await Promise.resolve();
+        expect(requests.filter((r) => r.method === "POST")).toHaveLength(1);
+    });
+
+    it("only DELETEs when the last subscriber for an id releases", async () => {
+        addHistoryViewerSubscription("hist-A");
+        addHistoryViewerSubscription("hist-A");
+        await Promise.resolve();
+        const postCount = requests.filter((r) => r.method === "POST").length;
+
+        removeHistoryViewerSubscription("hist-A");
+        await Promise.resolve();
+        // First remove still has one outstanding refcount — must not DELETE yet.
+        expect(requests.filter((r) => r.method === "DELETE")).toHaveLength(0);
+        expect(requests.filter((r) => r.method === "POST")).toHaveLength(postCount);
+
+        removeHistoryViewerSubscription("hist-A");
+        await Promise.resolve();
+        const deletes = requests.filter((r) => r.method === "DELETE");
+        expect(deletes).toHaveLength(1);
+        expect(deletes[0]?.history_ids).toEqual(["hist-A"]);
+    });
+
+    it("ignores unsubscribes for ids that were never subscribed", async () => {
+        removeHistoryViewerSubscription("hist-never");
+        await Promise.resolve();
+        expect(requests).toHaveLength(0);
+    });
+
+    it("tracks distinct history ids independently", async () => {
+        addHistoryViewerSubscription("hist-A");
+        addHistoryViewerSubscription("hist-B");
+        await Promise.resolve();
+        const ids = requests.filter((r) => r.method === "POST").map((r) => r.history_ids[0]);
+        expect(new Set(ids)).toEqual(new Set(["hist-A", "hist-B"]));
+    });
+});

client/src/composables/useNotificationSSE.ts

@@ -78,6 +78,11 @@ function openSourceIfNeeded() {
         // Global readiness flag so Selenium tests can distinguish a working
         // SSE pipeline from the polling fallback.
         sseGlobals().__galaxy_sse_connected = true;
+        // Re-assert any viewer subscriptions the user accumulated. The server
+        // doesn't carry app-level subscription state across reconnects (it
+        // only knows the user from the cookie), so the client owns the source
+        // of truth and replays it on every successful open.
+        replayViewerSubscriptionsOnOpen();
     };
 
     sharedSource.onerror = () => {
@@ -207,3 +212,82 @@ export function useSSEConnectionStatus() {
         hasEverConnected: readonly(sseEverConnected),
     };
 }
+
+// ---------------------------------------------------------------------------
+// Viewer subscriptions for non-owned histories
+//
+// Owner routing already covers history_update events for histories the
+// current user owns. Watching a non-owned history (e.g. a shared history
+// pinned in the multi-history view) requires the client to opt in by POSTing
+// the history id to ``/api/events/history-subscriptions``. The server keeps a
+// per-worker map and pushes the same history_update events to every viewer
+// in that map. The desired set is held module-level so reconnects can
+// replay it and so multiple components subscribed to the same id only emit
+// one HTTP call.
+// ---------------------------------------------------------------------------
+
+const viewerSubscriptions = new Map<string, number>();
+
+function postViewerSubscription(method: "POST" | "DELETE", historyIds: string[]): Promise<void> {
+    if (historyIds.length === 0) {
+        return Promise.resolve();
+    }
+    return fetch(withPrefix("/api/events/history-subscriptions"), {
+        method,
+        headers: { "Content-Type": "application/json", Accept: "application/json" },
+        credentials: "same-origin",
+        body: JSON.stringify({ history_ids: historyIds }),
+    }).then((response) => {
+        if (!response.ok) {
+            throw new Error(`history-subscriptions ${method} failed: ${response.status}`);
+        }
+    });
+}
+
+function replayViewerSubscriptionsOnOpen(): void {
+    const ids = [...viewerSubscriptions.keys()];
+    if (ids.length === 0) {
+        return;
+    }
+    postViewerSubscription("POST", ids).catch((err) =>
+        console.error("Failed to replay history viewer subscriptions on reconnect:", err),
+    );
+}
+
+/**
+ * Add a viewer subscription for a history this user does not own. Refcounted
+ * so two components watching the same history share one server-side
+ * subscription and the first to mount opens it, last to unmount closes it.
+ *
+ * Idempotent per call: re-subscribing an already-tracked id does not POST a
+ * duplicate.
+ */
+export function addHistoryViewerSubscription(historyId: string): void {
+    const previous = viewerSubscriptions.get(historyId) ?? 0;
+    viewerSubscriptions.set(historyId, previous + 1);
+    if (previous === 0) {
+        postViewerSubscription("POST", [historyId]).catch((err) =>
+            console.error(`Failed to subscribe to history ${historyId}:`, err),
+        );
+    }
+}
+
+export function removeHistoryViewerSubscription(historyId: string): void {
+    const previous = viewerSubscriptions.get(historyId);
+    if (!previous) {
+        return;
+    }
+    if (previous > 1) {
+        viewerSubscriptions.set(historyId, previous - 1);
+        return;
+    }
+    viewerSubscriptions.delete(historyId);
+    postViewerSubscription("DELETE", [historyId]).catch((err) =>
+        console.error(`Failed to unsubscribe from history ${historyId}:`, err),
+    );
+}
+
+/** Test-only: drain the desired set so per-test state doesn't leak. */
+export function _resetHistoryViewerSubscriptionsForTest(): void {
+    viewerSubscriptions.clear();
+}

lib/galaxy/managers/sse.py

@@ -85,6 +85,12 @@ def __init__(self, statsd_client: Optional[VanillaGalaxyStatsdClient] = None) ->
         self._broadcast_connections: set[asyncio.Queue] = set()
         self._loop: Optional[asyncio.AbstractEventLoop] = None
         self._statsd_client = statsd_client
+        # Viewer subscriptions for non-owned histories. Each worker keeps its
+        # own copy; the producer fans out subscribe/unsubscribe via Kombu so
+        # every webapp process learns about it. Maps encoded history_id (str)
+        # to the set of user_ids / session_ids that have asked for events.
+        self._history_viewers_user: dict[str, set[int]] = defaultdict(set)
+        self._history_viewers_session: dict[str, set[int]] = defaultdict(set)
 
     def _ensure_loop(self) -> None:
         """Capture the running asyncio event loop. Must be called from async context."""
@@ -134,10 +140,16 @@ def disconnect(
             self._connections[user_id].discard(queue)
             if not self._connections[user_id]:
                 del self._connections[user_id]
+                # Last connection for this user on this worker — drop their
+                # viewer subscriptions so the maps don't grow unbounded over
+                # long-lived processes. The client re-asserts subscriptions
+                # on every ``onopen`` so the next reconnect restores them.
+                self._purge_user_viewer_subscriptions(user_id)
         if galaxy_session_id is not None:
             self._session_connections[galaxy_session_id].discard(queue)
             if not self._session_connections[galaxy_session_id]:
                 del self._session_connections[galaxy_session_id]
+                self._purge_session_viewer_subscriptions(galaxy_session_id)
         self._broadcast_connections.discard(queue)
         log.debug(
             "SSE connection closed for user_id=%s session_id=%s (total=%d)",
@@ -167,6 +179,60 @@ def push_broadcast(self, event: SSEEvent) -> None:
         for queue in list(self._broadcast_connections):
             self._safe_put(queue, event)
 
+    def push_to_history_viewers(self, history_id: str, event: SSEEvent) -> None:
+        """Thread-safe. Push an event to every user/session that has subscribed
+        to ``history_id`` as a viewer (i.e. is watching a history they don't
+        own). Owner routing remains a separate path so the common "user changes
+        their own history" case never depends on a subscribe round trip.
+        """
+        for user_id in list(self._history_viewers_user.get(history_id, ())):
+            self.push_to_user(user_id, event)
+        for session_id in list(self._history_viewers_session.get(history_id, ())):
+            self.push_to_session(session_id, event)
+
+    # -- Viewer subscription bookkeeping --
+    # These mutate per-worker state and are called from the Kombu task thread
+    # via the ``subscribe_history_viewer`` / ``unsubscribe_history_viewer``
+    # control tasks. They are not thread-safe with each other (defaultdict
+    # mutations are not atomic), but Kombu drains the control queue serially
+    # so concurrent calls into a single manager don't happen in practice.
+
+    def subscribe_user_viewer(self, user_id: int, history_id: str) -> None:
+        self._history_viewers_user[history_id].add(user_id)
+
+    def unsubscribe_user_viewer(self, user_id: int, history_id: str) -> None:
+        viewers = self._history_viewers_user.get(history_id)
+        if viewers is None:
+            return
+        viewers.discard(user_id)
+        if not viewers:
+            del self._history_viewers_user[history_id]
+
+    def subscribe_session_viewer(self, session_id: int, history_id: str) -> None:
+        self._history_viewers_session[history_id].add(session_id)
+
+    def unsubscribe_session_viewer(self, session_id: int, history_id: str) -> None:
+        viewers = self._history_viewers_session.get(history_id)
+        if viewers is None:
+            return
+        viewers.discard(session_id)
+        if not viewers:
+            del self._history_viewers_session[history_id]
+
+    def _purge_user_viewer_subscriptions(self, user_id: int) -> None:
+        """Drop every viewer subscription this user holds on this worker.
+
+        Called after their last connection on this worker closes. The map of
+        ``history_id → user_ids`` is small enough that a full scan is fine —
+        the inverse index isn't worth maintaining at this scale.
+        """
+        for history_id in list(self._history_viewers_user.keys()):
+            self.unsubscribe_user_viewer(user_id, history_id)
+
+    def _purge_session_viewer_subscriptions(self, session_id: int) -> None:
+        for history_id in list(self._history_viewers_session.keys()):
+            self.unsubscribe_session_viewer(session_id, history_id)
+
     def _safe_put(self, queue: asyncio.Queue, event: SSEEvent) -> None:
         """Cross the thread boundary safely using ``call_soon_threadsafe``."""
         if self._loop is None or self._loop.is_closed():

lib/galaxy/managers/sse_dispatch.py

@@ -150,6 +150,39 @@ def history_update(
             kwargs["session_updates"] = session_updates
         self._send("history_update", kwargs)
 
+    def subscribe_history_viewer(
+        self,
+        history_id: str,
+        user_id: Optional[int] = None,
+        session_id: Optional[int] = None,
+    ) -> None:
+        """Broadcast a viewer-subscription record so every webapp worker can
+        push history_update events to a user/session watching a history they
+        don't own. The dispatch is fire-and-forget — workers apply the
+        record locally on receipt. Multi-worker deployments rely on this to
+        keep subscriptions consistent regardless of which worker fields the
+        REST call.
+        """
+        kwargs: dict[str, Any] = {"history_id": history_id}
+        if user_id is not None:
+            kwargs["user_id"] = user_id
+        if session_id is not None:
+            kwargs["session_id"] = session_id
+        self._send("subscribe_history_viewer", kwargs)
+
+    def unsubscribe_history_viewer(
+        self,
+        history_id: str,
+        user_id: Optional[int] = None,
+        session_id: Optional[int] = None,
+    ) -> None:
+        kwargs: dict[str, Any] = {"history_id": history_id}
+        if user_id is not None:
+            kwargs["user_id"] = user_id
+        if session_id is not None:
+            kwargs["session_id"] = session_id
+        self._send("unsubscribe_history_viewer", kwargs)
+
     def entry_point_update(self, user_id: int, event_id: Optional[str] = None) -> None:
         """Fan out a wake-up ``entry_point_update`` event for one user.