feat(frontend): useTailStream lifecycle hook with backoff + stall detection
This commit is contained in:
Tyler
@@ -0,0 +1,237 @@
|
||||
// ---------------------------------------------------------------------------
|
||||
// Live-tail connection lifecycle hook (sub-project 5, Phase 5 Task 19).
|
||||
//
|
||||
// Opens a `streamTail(resource)` connection, dispatches `item` events into
|
||||
// the matching `useTailStore` slice, exposes the connection status to the
|
||||
// page (for `<TailStatusPill>`), and survives transient backend failures
|
||||
// with an exponential-backoff retry (1s → 2s → 4s → 8s → 16s, capped at
|
||||
// 30s — spec §3.4). A stall detector flips status to `stalled` after 30s
|
||||
// of silence (no event, including heartbeats) so the UI can show a stale
|
||||
// connection without polling.
|
||||
//
|
||||
// Design notes:
|
||||
// - One effect per `(resource, reconnectNonce)` pair. Bumping
|
||||
// `reconnectNonce` from `forceReconnect()` re-runs the effect, which
|
||||
// aborts the in-flight stream (cleanup) and opens a fresh one.
|
||||
// - The AbortController is stored in a ref so `forceReconnect` can abort
|
||||
// even from outside React's render cycle.
|
||||
// - Backoff is indexed by an `attempt` counter that resets to 0 once the
|
||||
// server completes a snapshot (i.e. we're "live"). A clean server-side
|
||||
// EOF is treated like an error — we reconnect with backoff.
|
||||
// - `setStatus("closed")` fires from the effect cleanup; the consumer
|
||||
// (e.g. `<TailStatusPill>`) typically unmounts at the same time, so
|
||||
// this update is rarely observed — but it's there for any parent that
|
||||
// keeps the consumer rendered after the hook is detached.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
import { useCallback, useEffect, useRef, useState } from "react";
|
||||
import { streamTail, type TailResource } from "@/lib/tail-stream";
|
||||
import { useTailStore } from "@/store/tail-store";
|
||||
import type { Activity, Claim, Remittance } from "@/types";
|
||||
|
||||
export type TailStatus =
|
||||
| "connecting"
|
||||
| "live"
|
||||
| "reconnecting"
|
||||
| "closed"
|
||||
| "stalled"
|
||||
| "error";
|
||||
|
||||
export interface UseTailStreamResult {
|
||||
status: TailStatus;
|
||||
lastEventAt: Date | null;
|
||||
error: Error | null;
|
||||
forceReconnect: () => void;
|
||||
}
|
||||
|
||||
/** Backoff schedule per spec §3.4: 1s, 2s, 4s, 8s, 16s, then cap at 30s. */
|
||||
const BACKOFF_STEPS_MS: readonly number[] = [
|
||||
1_000, 2_000, 4_000, 8_000, 16_000, 30_000,
|
||||
];
|
||||
|
||||
/** No event (including heartbeat) for this long → flip to `stalled`. */
|
||||
const STALL_TIMEOUT_MS = 30_000;
|
||||
|
||||
function backoffDelayMs(attempt: number): number {
|
||||
const i = Math.min(Math.max(attempt, 0), BACKOFF_STEPS_MS.length - 1);
|
||||
return BACKOFF_STEPS_MS[i] as number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Dispatch a single item event into the matching slice of the tail store.
|
||||
* The store slices are typed with the canonical shapes (`Claim`,
|
||||
* `Remittance`, `Activity`); the stream yields `unknown` so we cast here.
|
||||
*/
|
||||
function dispatch(resource: TailResource, data: unknown): void {
|
||||
const store = useTailStore.getState();
|
||||
switch (resource) {
|
||||
case "claims":
|
||||
store.addClaim(data as Claim);
|
||||
break;
|
||||
case "remittances":
|
||||
store.addRemittance(data as Remittance);
|
||||
break;
|
||||
case "activity":
|
||||
store.addActivity(data as Activity);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
export function useTailStream(resource: TailResource): UseTailStreamResult {
|
||||
const [status, setStatus] = useState<TailStatus>("connecting");
|
||||
const [lastEventAt, setLastEventAt] = useState<Date | null>(null);
|
||||
const [error, setError] = useState<Error | null>(null);
|
||||
|
||||
/**
|
||||
* Bumping this state causes the effect to re-run, aborting the current
|
||||
* stream and starting a fresh one. `forceReconnect` is the only place
|
||||
* we mutate it from outside the effect itself.
|
||||
*/
|
||||
const [reconnectNonce, setReconnectNonce] = useState(0);
|
||||
|
||||
/**
|
||||
* Hold the in-flight AbortController so `forceReconnect` can tear it
|
||||
* down even when called from a stale render. The effect cleanup also
|
||||
* aborts via this ref.
|
||||
*/
|
||||
const abortRef = useRef<AbortController | null>(null);
|
||||
|
||||
const forceReconnect = useCallback(() => {
|
||||
setReconnectNonce((n) => n + 1);
|
||||
}, []);
|
||||
|
||||
useEffect(() => {
|
||||
let cancelled = false;
|
||||
let attempt = 0;
|
||||
let stallTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
let backoffTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
|
||||
const clearStall = (): void => {
|
||||
if (stallTimer) {
|
||||
clearTimeout(stallTimer);
|
||||
stallTimer = null;
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Re-arm the stall detector. Called on every event (including
|
||||
* heartbeats and `item_dropped` notices) so a quiet backend that
|
||||
* still pings every <30s doesn't get flagged as stalled. The stall
|
||||
* timer only flips status if we were already in a "still trying"
|
||||
* state — once we're `closed` or `error`, a missed heartbeat
|
||||
* shouldn't override that.
|
||||
*/
|
||||
const armStall = (): void => {
|
||||
clearStall();
|
||||
stallTimer = setTimeout(() => {
|
||||
if (cancelled) return;
|
||||
setStatus((prev) => {
|
||||
if (
|
||||
prev === "closed" ||
|
||||
prev === "reconnecting" ||
|
||||
prev === "error" ||
|
||||
prev === "stalled"
|
||||
) {
|
||||
return prev;
|
||||
}
|
||||
return "stalled";
|
||||
});
|
||||
}, STALL_TIMEOUT_MS);
|
||||
};
|
||||
|
||||
const scheduleReconnect = (): void => {
|
||||
if (cancelled) return;
|
||||
if (backoffTimer) {
|
||||
clearTimeout(backoffTimer);
|
||||
backoffTimer = null;
|
||||
}
|
||||
const delay = backoffDelayMs(attempt);
|
||||
backoffTimer = setTimeout(() => {
|
||||
if (cancelled) return;
|
||||
backoffTimer = null;
|
||||
attempt += 1;
|
||||
void openOnce();
|
||||
}, delay);
|
||||
};
|
||||
|
||||
const openOnce = async (): Promise<void> => {
|
||||
if (cancelled) return;
|
||||
const controller = new AbortController();
|
||||
abortRef.current = controller;
|
||||
// First attempt → "connecting". Subsequent retries → "reconnecting"
|
||||
// so the user can see that we've lost the prior connection.
|
||||
setStatus(attempt === 0 ? "connecting" : "reconnecting");
|
||||
|
||||
try {
|
||||
const iter = streamTail(resource, { signal: controller.signal });
|
||||
// eslint-disable-next-line @typescript-eslint/no-unused-vars
|
||||
for await (const ev of iter) {
|
||||
if (cancelled) return;
|
||||
setLastEventAt(new Date());
|
||||
armStall();
|
||||
|
||||
switch (ev.type) {
|
||||
case "snapshot_end":
|
||||
// Server finished replaying; we are now live. Reset the
|
||||
// backoff counter so a transient blip doesn't penalize the
|
||||
// next outage.
|
||||
attempt = 0;
|
||||
setStatus("live");
|
||||
break;
|
||||
case "item":
|
||||
dispatch(resource, ev.data);
|
||||
break;
|
||||
case "heartbeat":
|
||||
case "item_dropped":
|
||||
// No state change — the stall timer has already been
|
||||
// re-armed. These exist purely so the hook knows the
|
||||
// connection is alive.
|
||||
break;
|
||||
case "error":
|
||||
// Yielded (not thrown) error event. Promote to a thrown
|
||||
// Error so the catch block below runs the reconnect
|
||||
// machinery.
|
||||
throw new Error(ev.data.message);
|
||||
}
|
||||
}
|
||||
// Stream finished cleanly (server-side EOF). Treat as a
|
||||
// reconnect-trigger: if we're still mounted, schedule a retry.
|
||||
if (!cancelled) {
|
||||
setStatus("reconnecting");
|
||||
scheduleReconnect();
|
||||
}
|
||||
} catch (err) {
|
||||
if (cancelled) return;
|
||||
// An abort during teardown isn't really an error — the stream
|
||||
// is being torn down on purpose. Skip the reconnect.
|
||||
if (controller.signal.aborted) return;
|
||||
const e = err instanceof Error ? err : new Error(String(err));
|
||||
setError(e);
|
||||
setStatus("error");
|
||||
scheduleReconnect();
|
||||
}
|
||||
};
|
||||
|
||||
void openOnce();
|
||||
armStall();
|
||||
|
||||
return () => {
|
||||
cancelled = true;
|
||||
if (abortRef.current) {
|
||||
abortRef.current.abort();
|
||||
abortRef.current = null;
|
||||
}
|
||||
if (backoffTimer) {
|
||||
clearTimeout(backoffTimer);
|
||||
backoffTimer = null;
|
||||
}
|
||||
clearStall();
|
||||
// Per spec: on unmount, status = closed. This is the only way a
|
||||
// parent that keeps the consumer mounted (e.g. for a transition)
|
||||
// can observe the closed state.
|
||||
setStatus("closed");
|
||||
};
|
||||
}, [resource, reconnectNonce]);
|
||||
|
||||
return { status, lastEventAt, error, forceReconnect };
|
||||
}
|
||||
Reference in New Issue
Block a user