feat(msg-bus): resubscribe topics on reconnect
This commit is contained in:
@@ -6,6 +6,7 @@
|
|||||||
// `RtEventKind` — the switch below fails to type-check until every
|
// `RtEventKind` — the switch below fails to type-check until every
|
||||||
// arm is handled, keeping the FE exhaustive.
|
// arm is handled, keeping the FE exhaustive.
|
||||||
|
|
||||||
|
import { useReconnect } from './useReconnect.svelte';
|
||||||
import { useTopic } from './useTopic.svelte';
|
import { useTopic } from './useTopic.svelte';
|
||||||
import type RtEventParams from '$lib/generated/message-bus/RtEventParams';
|
import type RtEventParams from '$lib/generated/message-bus/RtEventParams';
|
||||||
import type RtRevokedParams from '$lib/generated/message-bus/RtRevokedParams';
|
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
|
/** Grant revoked or folder deleted — the subscription is gone
|
||||||
* server-side. Reasonable UX: toast + navigate away. */
|
* server-side. Reasonable UX: toast + navigate away. */
|
||||||
onRevoked?: (params: RtRevokedParams) => void;
|
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;
|
return id ? `folder:${id}` : null;
|
||||||
};
|
};
|
||||||
useTopic(topic, (params) => dispatch(params, handlers), handlers.onRevoked);
|
useTopic(topic, (params) => dispatch(params, handlers), handlers.onRevoked);
|
||||||
|
useReconnect(handlers.onReconnect);
|
||||||
}
|
}
|
||||||
|
|
||||||
function dispatch(params: RtEventParams, handlers: FolderTopicHandlers): void {
|
function dispatch(params: RtEventParams, handlers: FolderTopicHandlers): void {
|
||||||
|
|||||||
@@ -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();
|
||||||
|
});
|
||||||
|
}
|
||||||
@@ -66,6 +66,15 @@ export type EventHandler = (params: RtEventParams) => void;
|
|||||||
* fires this so the consumer can toast / redirect / whatever. */
|
* fires this so the consumer can toast / redirect / whatever. */
|
||||||
export type RevokedHandler = (params: RtRevokedParams) => void;
|
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
|
/** Handle returned by `subscribe`. Call to release one refcount on the
|
||||||
* topic; the client unsubscribes over the wire only when the last
|
* topic; the client unsubscribes over the wire only when the last
|
||||||
* refcount drops. Idempotent — calling twice from the same subscriber
|
* refcount drops. Idempotent — calling twice from the same subscriber
|
||||||
@@ -134,6 +143,17 @@ export class MessageBusClient {
|
|||||||
* `MAX_CONSECUTIVE_FAILURES` the client stops reconnecting and
|
* `MAX_CONSECUTIVE_FAILURES` the client stops reconnecting and
|
||||||
* requires an explicit `reconnect()` from the caller. */
|
* requires an explicit `reconnect()` from the caller. */
|
||||||
#consecutiveFailures = 0;
|
#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<ReconnectHandler>();
|
||||||
|
|
||||||
/** `topic` → `{count, handlers, revokedHandlers, acked}`. Refcount
|
/** `topic` → `{count, handlers, revokedHandlers, acked}`. Refcount
|
||||||
* drives the wire: first refcount ⇒ send `rt.subscribe`; last drop
|
* 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
|
/** Force a fresh reconnect — for a live-updates toggle or a manual
|
||||||
* "reconnect" button. Rare; not part of the normal flow. Also the
|
* "reconnect" button. Rare; not part of the normal flow. Also the
|
||||||
* escape hatch after the circuit breaker trips: zeroes the
|
* escape hatch after the circuit breaker trips: zeroes the
|
||||||
@@ -320,6 +369,10 @@ export class MessageBusClient {
|
|||||||
this.state = 'connected';
|
this.state = 'connected';
|
||||||
this.#backoffMs = RECONNECT_MIN_MS;
|
this.#backoffMs = RECONNECT_MIN_MS;
|
||||||
this.#consecutiveFailures = 0;
|
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
|
// Replay every already-known topic. `entry.acked` is reset here
|
||||||
// because the fresh connection has no server-side memory of
|
// because the fresh connection has no server-side memory of
|
||||||
// prior subscriptions.
|
// prior subscriptions.
|
||||||
@@ -329,6 +382,24 @@ export class MessageBusClient {
|
|||||||
busLog.warn('resubscribe failed', { topic, error: err })
|
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 {
|
#onMessage(ev: MessageEvent): void {
|
||||||
|
|||||||
@@ -484,6 +484,21 @@
|
|||||||
);
|
);
|
||||||
busLog.warn('folder access revoked', { topic: params.topic, reason: params.reason });
|
busLog.warn('folder access revoked', { topic: params.topic, reason: params.reason });
|
||||||
void goto(resolve('/files'));
|
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('');
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user