diff --git a/frontend/src/lib/composables/useFolderTopic.svelte.ts b/frontend/src/lib/composables/useFolderTopic.svelte.ts index 747bec37..9193fb56 100644 --- a/frontend/src/lib/composables/useFolderTopic.svelte.ts +++ b/frontend/src/lib/composables/useFolderTopic.svelte.ts @@ -6,6 +6,7 @@ // `RtEventKind` — the switch below fails to type-check until every // arm is handled, keeping the FE exhaustive. +import { useReconnect } from './useReconnect.svelte'; import { useTopic } from './useTopic.svelte'; import type RtEventParams from '$lib/generated/message-bus/RtEventParams'; import type RtRevokedParams from '$lib/generated/message-bus/RtRevokedParams'; @@ -41,6 +42,13 @@ export interface FolderTopicHandlers { /** Grant revoked or folder deleted — the subscription is gone * server-side. Reasonable UX: toast + navigate away. */ onRevoked?: (params: RtRevokedParams) => void; + /** WS reconnected after a prior disconnect. Bus events published + * during the outage window are lost (in-memory bus, no replay), + * so the folder view has to refetch to catch up with the server. + * Typical wiring: `onReconnect: () => reload()`. Not called on + * the initial connect — the caller's own load path handles that. + * See `project_message_bus_reconnect_gap` memory. */ + onReconnect?: () => void; } /** @@ -61,6 +69,7 @@ export function useFolderTopic( return id ? `folder:${id}` : null; }; useTopic(topic, (params) => dispatch(params, handlers), handlers.onRevoked); + useReconnect(handlers.onReconnect); } function dispatch(params: RtEventParams, handlers: FolderTopicHandlers): void { diff --git a/frontend/src/lib/composables/useReconnect.svelte.ts b/frontend/src/lib/composables/useReconnect.svelte.ts new file mode 100644 index 00000000..476b761a --- /dev/null +++ b/frontend/src/lib/composables/useReconnect.svelte.ts @@ -0,0 +1,27 @@ +// Svelte 5 rune wrapper around `messageBus.onReconnect`. +// +// Fires the given callback the first time the WS reconnects after a +// prior disconnect (server restart, network blip, sleep/wake). +// **Not** called on the initial connect — the caller's own load path +// is already fetching then. Bridges the "events published during the +// disconnect window are lost" gap; consumers typically pass +// `reload()` so the view catches up with the server after the outage. +// +// See `client.svelte.ts::onReconnect` for lifecycle details and +// `project_message_bus_reconnect_gap` memory for the gap it closes. + +import { messageBus } from '$lib/message-bus/client.svelte'; + +/** + * Register `cb` as a reconnect handler for the lifetime of the + * calling component. Auto-unregisters on destroy via `$effect` + * cleanup. Passing `null`/`undefined` is a no-op — convenient for + * conditional wiring (`useReconnect(handlers.onReconnect)`). + */ +export function useReconnect(cb: (() => void) | null | undefined): void { + $effect(() => { + if (!cb) return; + const release = messageBus.onReconnect(cb); + return () => release(); + }); +} diff --git a/frontend/src/lib/message-bus/client.svelte.ts b/frontend/src/lib/message-bus/client.svelte.ts index a381ae30..fcc8982a 100644 --- a/frontend/src/lib/message-bus/client.svelte.ts +++ b/frontend/src/lib/message-bus/client.svelte.ts @@ -66,6 +66,15 @@ export type EventHandler = (params: RtEventParams) => void; * fires this so the consumer can toast / redirect / whatever. */ export type RevokedHandler = (params: RtRevokedParams) => void; +/** Callback invoked when the WS reconnects AFTER a prior disconnect — + * never on the first connect. Fires after client-side sub replay has + * been kicked off (`#sendSubscribe` for every known topic), so the + * handler can safely call `reload()`-style refetches knowing the + * post-reconnect event stream is armed. Bridges the "events published + * during the disconnect window are lost" gap — see + * `project_message_bus_reconnect_gap` memory. */ +export type ReconnectHandler = () => void; + /** Handle returned by `subscribe`. Call to release one refcount on the * topic; the client unsubscribes over the wire only when the last * refcount drops. Idempotent — calling twice from the same subscriber @@ -134,6 +143,17 @@ export class MessageBusClient { * `MAX_CONSECUTIVE_FAILURES` the client stops reconnecting and * requires an explicit `reconnect()` from the caller. */ #consecutiveFailures = 0; + /** True once we've observed at least one successful `#onOpen`. + * Used to distinguish "initial connect" (don't fire onReconnect + * handlers — the initial load path is doing the fetch already) + * from "reconnect" (do fire — events during the outage window + * were lost, consumers must refetch). */ + #hasConnectedBefore = false; + /** Reconnect handlers, invoked from `#onOpen` on the SECOND-onwards + * successful connect. Plain Set — internal registry, not + * reactive. Same rationale as `#subs` / `#pending`. */ + // eslint-disable-next-line svelte/prefer-svelte-reactivity + #reconnectHandlers = new Set(); /** `topic` → `{count, handlers, revokedHandlers, acked}`. Refcount * drives the wire: first refcount ⇒ send `rt.subscribe`; last drop @@ -225,6 +245,35 @@ export class MessageBusClient { }); } + /** + * Register a handler that fires when the WS reconnects AFTER a + * prior disconnect (server restart, network blip, sleep/wake). + * NOT called on the initial connect — that path is already + * handled by the consumer's own load logic. Returns an + * unsubscribe fn. + * + * Wrapped in `untrack` for the same reason `subscribe` is — + * reading `#hasConnectedBefore` etc. inside a caller's `$effect` + * would leak a reactive dep. Callers reach for this via the + * `useReconnect` composable, which manages the lifecycle. + * + * Bridges the "events lost during outage window" gap: consumers + * refetch on reconnect to bring their view back in line with the + * server, since bus publishes during the disconnect never reached + * this session. See `project_message_bus_reconnect_gap` memory. + */ + onReconnect(cb: ReconnectHandler): () => void { + return untrack(() => { + this.#reconnectHandlers.add(cb); + let released = false; + return () => { + if (released) return; + released = true; + this.#reconnectHandlers.delete(cb); + }; + }); + } + /** Force a fresh reconnect — for a live-updates toggle or a manual * "reconnect" button. Rare; not part of the normal flow. Also the * escape hatch after the circuit breaker trips: zeroes the @@ -320,6 +369,10 @@ export class MessageBusClient { this.state = 'connected'; this.#backoffMs = RECONNECT_MIN_MS; this.#consecutiveFailures = 0; + // Snapshot whether this is a reconnect BEFORE we flip the + // `hasConnectedBefore` bit, so handlers only fire on 2nd+ open. + const isReconnect = this.#hasConnectedBefore; + this.#hasConnectedBefore = true; // Replay every already-known topic. `entry.acked` is reset here // because the fresh connection has no server-side memory of // prior subscriptions. @@ -329,6 +382,24 @@ export class MessageBusClient { busLog.warn('resubscribe failed', { topic, error: err }) ); } + // Fire reconnect handlers AFTER sub replay is kicked (the + // `rt.subscribe` frames are on the socket; ack may be + // in-flight). Handlers refetching state via REST will see a + // consistent post-reconnect view; any events published between + // resubscribe and the handler's refetch race safely — a stale + // event just means one extra `reload()` on the next tick. + if (isReconnect && this.#reconnectHandlers.size > 0) { + busLog.debug('firing reconnect handlers', { + count: this.#reconnectHandlers.size + }); + for (const cb of this.#reconnectHandlers) { + try { + cb(); + } catch (err) { + busLog.warn('reconnect handler threw', { error: err }); + } + } + } } #onMessage(ev: MessageEvent): void { diff --git a/frontend/src/routes/files/[...path]/+page.svelte b/frontend/src/routes/files/[...path]/+page.svelte index 3b4cf19b..9db4cf26 100644 --- a/frontend/src/routes/files/[...path]/+page.svelte +++ b/frontend/src/routes/files/[...path]/+page.svelte @@ -484,6 +484,21 @@ ); busLog.warn('folder access revoked', { topic: params.topic, reason: params.reason }); void goto(resolve('/files')); + }, + onReconnect: () => { + // WS reconnected after a prior disconnect — any bus events + // published during the outage window were dropped by the + // in-memory bus (no replay). Force a refetch so the listing + // catches up with the server-authoritative state. Goes + // through the same `scheduleLiveReload` coalescer as event- + // driven refreshes so a burst of reconnects (rare, but the + // circuit breaker can produce one) collapses to a single + // fetch. Passing an actor of `null`-equivalent — use an + // empty string so the echo-skip's `actor === user.id` + // check never matches. See + // `project_message_bus_reconnect_gap` memory. + busLog.warn('reconnected — refetching folder'); + scheduleLiveReload(''); } });