From bf918d1a117b9acf0646faca10d9cb4f5bd4f770 Mon Sep 17 00:00:00 2001 From: Hare Date: Sat, 3 Oct 2026 23:35:43 +0900 Subject: [PATCH] fix: route sidebar failures through workspace alerts --- .../src/lib/workspace/alerts/store.ts | 8 + .../console/worker-console.ui.test.ts | 2 +- .../lib/workspace/multiplexer.browser.spec.ts | 90 +++++- .../src/lib/workspace/multiplexer.ts | 145 ++++++--- .../sidebar/GlobalNavSections.svelte | 4 - .../sidebar/WorkersNavSection.svelte | 7 +- .../workspace/sidebar/WorkspaceSidebar.svelte | 4 +- .../sidebar/WorkspaceSwitcher.svelte | 26 +- .../worker-subscription.browser.spec.ts | 303 ++++++++++++++++++ .../workspace/sidebar/worker-subscription.ts | 194 ++++++++--- .../workspace-switcher.browser.spec.ts | 70 ++++ web/workspace/src/routes/+layout.svelte | 13 + .../src/routes/sidebar-fold.browser.spec.ts | 17 + .../src/routes/w/[workspaceId]/+layout.svelte | 24 +- .../workers/[workerRef]/console/+page.svelte | 6 +- .../console/console-display.browser.spec.ts | 33 +- ...workspace-layout-lifecycle.browser.spec.ts | 91 ++++++ 17 files changed, 900 insertions(+), 137 deletions(-) create mode 100644 web/workspace/src/lib/workspace/sidebar/worker-subscription.browser.spec.ts create mode 100644 web/workspace/src/lib/workspace/sidebar/workspace-switcher.browser.spec.ts create mode 100644 web/workspace/src/routes/w/[workspaceId]/workspace-layout-lifecycle.browser.spec.ts diff --git a/web/workspace/src/lib/workspace/alerts/store.ts b/web/workspace/src/lib/workspace/alerts/store.ts index a05384315..5623b76b6 100644 --- a/web/workspace/src/lib/workspace/alerts/store.ts +++ b/web/workspace/src/lib/workspace/alerts/store.ts @@ -40,6 +40,14 @@ export function pushWorkspaceAlert( ): string { const id = options.id ?? `${Date.now().toString(36)}-${(sequence++).toString(36)}`; + const existing = alerts.find((alert) => alert.id === id); + if ( + existing?.level === level && + existing.title === options.title && + existing.message === message + ) { + return id; + } alerts = [ ...alerts.filter((alert) => alert.id !== id), { diff --git a/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts b/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts index 936c7501f..ba0543d4d 100644 --- a/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts +++ b/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts @@ -1438,7 +1438,7 @@ Deno.test("Workspace Worker list and Console share the multiplexed connection", consolePage.includes("workspaceMultiplexer(target.workspaceId)") && sidebarStore.includes("workspaceMultiplexer(workspaceId)") && multiplexer.includes("const multiplexers = new Map") && - multiplexer.includes("frame: 'worker_protocol'"), + multiplexer.includes('frame: "worker_protocol"'), "Sidebar and Console should share one Workspace multiplexer and route Worker methods through a subscription lane", ); assert( diff --git a/web/workspace/src/lib/workspace/multiplexer.browser.spec.ts b/web/workspace/src/lib/workspace/multiplexer.browser.spec.ts index 251ebe0fc..cc89a964d 100644 --- a/web/workspace/src/lib/workspace/multiplexer.browser.spec.ts +++ b/web/workspace/src/lib/workspace/multiplexer.browser.spec.ts @@ -99,7 +99,10 @@ test('invalid inbound frame closes before dispatch and reconnects for a fresh sn expect(onFrame).not.toHaveBeenCalled(); expect(first.closes).toEqual([[4002, 'Invalid workspace protocol frame']]); - expect(onStatus).toHaveBeenLastCalledWith('closed', 'Workspace protocol frame rejected'); + expect(onStatus).toHaveBeenLastCalledWith('closed', { + kind: 'protocol', + message: 'Workspace protocol frame rejected', + }); expect(JSON.stringify(onStatus.mock.calls)).not.toContain('untrusted detail'); vi.advanceTimersByTime(500); @@ -108,9 +111,84 @@ test('invalid inbound frame closes before dispatch and reconnects for a fresh sn second.open(); expect(second.sent).toHaveLength(1); expect(JSON.parse(second.sent[0]).message.method).toBe('subscribe_events'); + + first.open(); + first.message('{"protocol_version":2,"frame":"invalid"}'); + expect(second.sent).toHaveLength(1); + expect(second.closes).toEqual([]); multiplexer.dispose(); }); +test('normal disposal closes silently and never schedules a reconnect', () => { + vi.useFakeTimers(); + const multiplexer = new WorkspaceMultiplexer('workspace-dispose-test'); + const onStatus = vi.fn(); + multiplexer.subscribe( + { topic: 'workspace_workers' }, + { onFrame: vi.fn(), onStatus }, + ); + + const socket = FakeWebSocket.instances[0]; + socket.open(); + onStatus.mockClear(); + multiplexer.dispose(); + + expect(socket.closes).toEqual([[undefined, undefined]]); + expect(onStatus).not.toHaveBeenCalled(); + vi.advanceTimersByTime(1_000); + expect(FakeWebSocket.instances).toHaveLength(1); +}); + +test('terminal subscription closure does not retry or report normal socket cleanup as transport failure', () => { + vi.useFakeTimers(); + const multiplexer = new WorkspaceMultiplexer('workspace-terminal-test'); + const onFrame = vi.fn(); + const onStatus = vi.fn(); + multiplexer.subscribe( + { topic: 'workspace_workers' }, + { onFrame, onStatus }, + ); + + const socket = FakeWebSocket.instances[0]; + socket.open(); + const request = JSON.parse(socket.sent[0]); + socket.message(JSON.stringify({ + protocol_version: 2, + frame: 'response', + message: { + result: 'subscribed', + payload: { + request_id: request.message.params.request_id, + subscription_id: 'subscription-terminal', + selector: { topic: 'workspace_workers' }, + snapshot: { topic: 'workers', data: { workers: [] } }, + }, + }, + })); + onStatus.mockClear(); + const sentBeforeClosure = socket.sent.length; + + socket.message(JSON.stringify({ + protocol_version: 2, + frame: 'event', + message: { + event: 'subscription_closed', + data: { + subscription_id: 'subscription-terminal', + code: 'unauthorized', + message: 'Workspace Worker subscription is no longer authorized', + }, + }, + })); + + expect(onFrame).toHaveBeenCalledTimes(2); + expect(onStatus).not.toHaveBeenCalled(); + expect(socket.sent).toHaveLength(sentBeforeClosure); + expect(socket.closes).toEqual([[undefined, undefined]]); + vi.advanceTimersByTime(1_000); + expect(FakeWebSocket.instances).toHaveLength(1); +}); + test('valid subscription closure remains isolated and resubscribes only its selector', () => { const multiplexer = new WorkspaceMultiplexer('workspace-isolation-test'); const workersFrame = vi.fn(); @@ -173,15 +251,9 @@ test('valid subscription closure remains isolated and resubscribes only its sele ); expect(workersFrame).toHaveBeenCalledTimes(2); - expect(workersStatus).toHaveBeenLastCalledWith( - 'connecting', - 'resubscribe for a fresh snapshot', - ); + expect(workersStatus).toHaveBeenLastCalledWith('connecting'); expect(workdirsFrame).toHaveBeenCalledTimes(1); - expect(workdirsStatus).not.toHaveBeenCalledWith( - 'connecting', - 'resubscribe for a fresh snapshot', - ); + expect(workdirsStatus).not.toHaveBeenCalledWith('connecting'); expect(socket.sent).toHaveLength(sentBeforeClosure + 1); expect(JSON.parse(socket.sent.at(-1)!).message.params.selector).toEqual({ topic: 'workspace_workers', diff --git a/web/workspace/src/lib/workspace/multiplexer.ts b/web/workspace/src/lib/workspace/multiplexer.ts index 17aacbaf7..0e2007dcb 100644 --- a/web/workspace/src/lib/workspace/multiplexer.ts +++ b/web/workspace/src/lib/workspace/multiplexer.ts @@ -1,19 +1,27 @@ -import { browser } from '$app/env'; +import { browser } from "$app/env"; import type { EventSubscriptionSelector, Method, SubscriptionFrame, SubscriptionId, -} from '#lib/generated/protocol.ts'; +} from "#lib/generated/protocol.ts"; import { decodeSubscriptionFrame, subscriptionFrameMatchesSelector, -} from '#lib/generated/protocol-validator.ts'; -import { workspaceApiPath } from '#lib/workspace/api/http.ts'; +} from "#lib/generated/protocol-validator.ts"; +import { workspaceApiPath } from "#lib/workspace/api/http.ts"; + +export type WorkspaceMultiplexerFailure = { + kind: "protocol" | "transport"; + message: string; +}; type Listener = { onFrame(frame: SubscriptionFrame): void; - onStatus?(status: 'connecting' | 'open' | 'closed', message?: string): void; + onStatus?( + status: "connecting" | "open" | "closed", + failure?: WorkspaceMultiplexerFailure, + ): void; }; type ActiveSubscription = { @@ -33,12 +41,14 @@ const multiplexers = new Map(); const INVALID_PROTOCOL_CLOSE_CODE = 4002; let nextMultiplexerSequence = 0; -function nextMultiplexerId(kind: 'client' | 'request'): string { +function nextMultiplexerId(kind: "client" | "request"): string { nextMultiplexerSequence += 1; return `${kind}-${nextMultiplexerSequence}`; } -export function workspaceMultiplexer(workspaceId: string): WorkspaceMultiplexer { +export function workspaceMultiplexer( + workspaceId: string, +): WorkspaceMultiplexer { let multiplexer = multiplexers.get(workspaceId); if (!multiplexer) { multiplexer = new WorkspaceMultiplexer(workspaceId); @@ -61,7 +71,7 @@ export class WorkspaceMultiplexer { readonly #runtimeSubscriptions = new Map(); #socket: WebSocket | null = null; #reconnectTimer: ReturnType | null = null; - #disconnectMessage: string | null = null; + #disconnectFailure: WorkspaceMultiplexerFailure | null = null; #closed = false; constructor(workspaceId: string) { @@ -72,7 +82,7 @@ export class WorkspaceMultiplexer { selector: EventSubscriptionSelector, listener: Listener, ): WorkspaceMultiplexerSubscription { - const clientId = nextMultiplexerId('client'); + const clientId = nextMultiplexerId("client"); const subscription: ActiveSubscription = { clientId, selector, @@ -83,7 +93,7 @@ export class WorkspaceMultiplexer { this.#subscriptions.set(clientId, subscription); this.#closed = false; if (this.#socket?.readyState === WebSocket.OPEN) { - subscription.listener.onStatus?.('open'); + subscription.listener.onStatus?.("open"); this.#sendSubscribe(subscription); } else { this.#ensureConnected(); @@ -95,37 +105,48 @@ export class WorkspaceMultiplexer { } #ensureConnected(): void { - if (!browser || this.#socket || this.#closed || this.#subscriptions.size === 0) return; + if ( + !browser || this.#socket || this.#closed || this.#subscriptions.size === 0 + ) return; for (const subscription of this.#subscriptions.values()) { - subscription.listener.onStatus?.('connecting'); + subscription.listener.onStatus?.("connecting"); } const url = new URL( - workspaceApiPath(this.#workspaceId, '/protocol/ws'), + workspaceApiPath(this.#workspaceId, "/protocol/ws"), window.location.origin, ); - url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'; + url.protocol = url.protocol === "https:" ? "wss:" : "ws:"; const socket = new WebSocket(url); this.#socket = socket; - socket.addEventListener('open', () => { + socket.addEventListener("open", () => { + if (this.#socket !== socket) return; for (const subscription of this.#subscriptions.values()) { - subscription.listener.onStatus?.('open'); + subscription.listener.onStatus?.("open"); this.#sendSubscribe(subscription); } }); - socket.addEventListener('message', (event) => this.#receive(event.data)); - socket.addEventListener('error', () => socket.close()); - socket.addEventListener('close', () => { + socket.addEventListener("message", (event) => { + if (this.#socket !== socket) return; + this.#receive(event.data); + }); + socket.addEventListener("error", () => { + if (this.#socket !== socket) return; + socket.close(); + }); + socket.addEventListener("close", () => { if (this.#socket !== socket) return; this.#socket = null; - const disconnectMessage = - this.#disconnectMessage ?? 'Workspace subscription disconnected'; - this.#disconnectMessage = null; + const failure = this.#disconnectFailure ?? { + kind: "transport" as const, + message: "Workspace subscription disconnected", + }; + this.#disconnectFailure = null; this.#requests.clear(); this.#runtimeSubscriptions.clear(); for (const subscription of this.#subscriptions.values()) { subscription.requestId = null; subscription.subscriptionId = null; - subscription.listener.onStatus?.('closed', disconnectMessage); + subscription.listener.onStatus?.("closed", failure); } if (!this.#closed && this.#subscriptions.size > 0) { this.#reconnectTimer = setTimeout(() => this.#ensureConnected(), 500); @@ -134,14 +155,14 @@ export class WorkspaceMultiplexer { } #sendSubscribe(subscription: ActiveSubscription): void { - const requestId = nextMultiplexerId('request'); + const requestId = nextMultiplexerId("request"); subscription.requestId = requestId; this.#requests.set(requestId, subscription.clientId); this.#send({ protocol_version: 2, - frame: 'request', + frame: "request", message: { - method: 'subscribe_events', + method: "subscribe_events", params: { request_id: requestId, selector: subscription.selector }, }, }); @@ -154,9 +175,11 @@ export class WorkspaceMultiplexer { return; } const frame = decoded.value; - if (frame.frame === 'response' && frame.message.result === 'subscribed') { + if (frame.frame === "response" && frame.message.result === "subscribed") { const clientId = this.#requests.get(frame.message.payload.request_id); - const subscription = clientId ? this.#subscriptions.get(clientId) : undefined; + const subscription = clientId + ? this.#subscriptions.get(clientId) + : undefined; if (!clientId || !subscription) return; if (!subscriptionFrameMatchesSelector(frame, subscription.selector)) { this.#rejectInboundFrame(); @@ -169,9 +192,14 @@ export class WorkspaceMultiplexer { subscription.listener.onFrame(frame); return; } - if (frame.frame === 'response' && frame.message.result === 'subscription_rejected') { + if ( + frame.frame === "response" && + frame.message.result === "subscription_rejected" + ) { const clientId = this.#requests.get(frame.message.payload.request_id); - const subscription = clientId ? this.#subscriptions.get(clientId) : undefined; + const subscription = clientId + ? this.#subscriptions.get(clientId) + : undefined; subscription?.listener.onFrame(frame); if (clientId) { this.#requests.delete(frame.message.payload.request_id); @@ -179,9 +207,13 @@ export class WorkspaceMultiplexer { } return; } - if (frame.frame === 'event') { - const clientId = this.#runtimeSubscriptions.get(frame.message.data.subscription_id); - const subscription = clientId ? this.#subscriptions.get(clientId) : undefined; + if (frame.frame === "event") { + const clientId = this.#runtimeSubscriptions.get( + frame.message.data.subscription_id, + ); + const subscription = clientId + ? this.#subscriptions.get(clientId) + : undefined; if ( subscription && !subscriptionFrameMatchesSelector(frame, subscription.selector) @@ -191,30 +223,45 @@ export class WorkspaceMultiplexer { } subscription?.listener.onFrame(frame); if ( - frame.message.event === 'subscription_closed' && + frame.message.event === "subscription_closed" && clientId && - subscription && - this.#socket?.readyState === WebSocket.OPEN + subscription ) { this.#runtimeSubscriptions.delete(frame.message.data.subscription_id); subscription.subscriptionId = null; - subscription.listener.onStatus?.('connecting', frame.message.data.message); - this.#sendSubscribe(subscription); + if ( + (frame.message.data.code === "lagged" || + frame.message.data.code === "server_shutdown") && + this.#socket?.readyState === WebSocket.OPEN + ) { + subscription.listener.onStatus?.("connecting"); + this.#sendSubscribe(subscription); + } else { + this.#remove(clientId); + } } } } #rejectInboundFrame(): void { - this.#disconnectMessage = 'Workspace protocol frame rejected'; - this.#socket?.close(INVALID_PROTOCOL_CLOSE_CODE, 'Invalid workspace protocol frame'); + this.#disconnectFailure = { + kind: "protocol", + message: "Workspace protocol frame rejected", + }; + this.#socket?.close( + INVALID_PROTOCOL_CLOSE_CODE, + "Invalid workspace protocol frame", + ); } #sendWorkerMethod(clientId: string, method: Method): void { const subscription = this.#subscriptions.get(clientId); - if (!subscription?.subscriptionId) throw new Error('Worker protocol subscription is not open'); + if (!subscription?.subscriptionId) { + throw new Error("Worker protocol subscription is not open"); + } this.#send({ protocol_version: 2, - frame: 'worker_protocol', + frame: "worker_protocol", message: { subscription_id: subscription.subscriptionId, method }, }); } @@ -223,14 +270,16 @@ export class WorkspaceMultiplexer { const subscription = this.#subscriptions.get(clientId); if (!subscription) return; this.#subscriptions.delete(clientId); - if (subscription.subscriptionId && this.#socket?.readyState === WebSocket.OPEN) { + if ( + subscription.subscriptionId && this.#socket?.readyState === WebSocket.OPEN + ) { this.#send({ protocol_version: 2, - frame: 'request', + frame: "request", message: { - method: 'unsubscribe_events', + method: "unsubscribe_events", params: { - request_id: nextMultiplexerId('request'), + request_id: nextMultiplexerId("request"), subscription_id: subscription.subscriptionId, }, }, @@ -251,12 +300,10 @@ export class WorkspaceMultiplexer { clearTimeout(this.#reconnectTimer); this.#reconnectTimer = null; } - for (const subscription of this.#subscriptions.values()) { - subscription.listener.onStatus?.('closed', 'Workspace selection changed'); - } this.#subscriptions.clear(); this.#requests.clear(); this.#runtimeSubscriptions.clear(); + this.#disconnectFailure = null; this.#socket?.close(); this.#socket = null; } diff --git a/web/workspace/src/lib/workspace/sidebar/GlobalNavSections.svelte b/web/workspace/src/lib/workspace/sidebar/GlobalNavSections.svelte index 60527f6ab..b6deb6d9e 100644 --- a/web/workspace/src/lib/workspace/sidebar/GlobalNavSections.svelte +++ b/web/workspace/src/lib/workspace/sidebar/GlobalNavSections.svelte @@ -44,10 +44,6 @@