diff --git a/src/Web/autopilot-monitor-web/app/progress/hooks/useProgressEvents.ts b/src/Web/autopilot-monitor-web/app/progress/hooks/useProgressEvents.ts index f5ffb94d..841ec756 100644 --- a/src/Web/autopilot-monitor-web/app/progress/hooks/useProgressEvents.ts +++ b/src/Web/autopilot-monitor-web/app/progress/hooks/useProgressEvents.ts @@ -3,9 +3,15 @@ import { useCallback, useEffect, useRef, useState } from "react"; import { api } from "@/lib/api"; import { authenticatedFetch, TokenExpiredError } from "@/lib/authenticatedFetch"; +import { createBurstScheduler, type BurstScheduler } from "@/lib/burstScheduler"; import { EnrollmentEvent, Session } from "@/types"; import type { NotificationType } from "@/contexts/NotificationContext"; +// Live refetch coalescing on SignalR signals (see lib/burstScheduler). Each refresh is two +// requests (summary lookup + events), hence a slightly wider window than the session page. +const REFRESH_TRAILING_MS = 500; +const REFRESH_MAX_WAIT_MS = 1_500; + type AddNotification = ( type: NotificationType, title: string, @@ -26,14 +32,15 @@ export interface UseProgressEventsReturn { events: EnrollmentEvent[]; setEvents: React.Dispatch>; sessionRef: React.RefObject; - scheduleFetchEvents: (delayMs?: number) => void; + /** Coalesced live refetch — call on every SignalR signal, the scheduler decides when to fetch. */ + scheduleFetchEvents: () => void; } /** * Owns the progress page's event list lifecycle: - * - keeps a sessionRef in sync (used by SignalR hook + debounced refetch) + * - keeps a sessionRef in sync (used by SignalR hook + coalesced refetch) * - initial event fetch once per session (StrictMode-safe guard) - * - debounced scheduleFetchEvents that refreshes session summary + events + * - burst-coalesced scheduleFetchEvents that refreshes session summary + events * - merge-by-sequence dedup so repeat signals don't duplicate rows */ export function useProgressEvents({ @@ -46,7 +53,10 @@ export function useProgressEvents({ const [events, setEvents] = useState([]); const sessionRef = useRef(null); const lastFetchedSessionId = useRef(null); - const refetchTimerRef = useRef | null>(null); + // The scheduler is created lazily on the first signal (refs stay out of render) and + // runs through refreshRef so it always sees the latest closure. + const refreshRef = useRef<() => Promise>(async () => {}); + const refreshScheduler = useRef(null); useEffect(() => { sessionRef.current = session; @@ -114,57 +124,65 @@ export function useProgressEvents({ void run(); }, [session, tenantId, getAccessToken, addNotification]); - const scheduleFetchEvents = useCallback( - (delayMs: number = 500) => { - if (refetchTimerRef.current) clearTimeout(refetchTimerRef.current); - refetchTimerRef.current = setTimeout(async () => { - const currentSession = sessionRef.current; - if (!currentSession) return; - try { - // Refresh the session summary via the serial lookup (the canonical serial from the - // session row always matches exactly). Guard on sessionId: the lookup returns the - // NEWEST session for the serial, which after a re-enrollment is a different session — - // the page must keep showing the one the user selected. - const lookupResponse = await authenticatedFetch( - api.progress.lookup(tenantId, currentSession.serialNumber), - getAccessToken, - ); - if (lookupResponse.ok) { - const lookupData = await lookupResponse.json(); - const updated: Session | null = lookupData.found ? lookupData.session : null; - if (updated && updated.sessionId === currentSession.sessionId) { - setSession(updated); - } + const refresh = useCallback( + async () => { + const currentSession = sessionRef.current; + if (!currentSession) return; + try { + // Refresh the session summary via the serial lookup (the canonical serial from the + // session row always matches exactly). Guard on sessionId: the lookup returns the + // NEWEST session for the serial, which after a re-enrollment is a different session — + // the page must keep showing the one the user selected. + const lookupResponse = await authenticatedFetch( + api.progress.lookup(tenantId, currentSession.serialNumber), + getAccessToken, + ); + if (lookupResponse.ok) { + const lookupData = await lookupResponse.json(); + const updated: Session | null = lookupData.found ? lookupData.session : null; + if (updated && updated.sessionId === currentSession.sessionId) { + setSession(updated); } - - const eventsResponse = await authenticatedFetch( - api.progress.sessionEvents(currentSession.sessionId, tenantId, currentSession.serialNumber), - getAccessToken, - ); - if (eventsResponse.ok) { - const eventsData = await eventsResponse.json(); - const fetched: EnrollmentEvent[] = eventsData.events || []; - setEvents((prev) => { - const existingIds = new Set(prev.map((e) => e.eventId)); - const newEvents = fetched.filter((e) => !existingIds.has(e.eventId)); - if (newEvents.length === 0) return prev; - return [...prev, ...newEvents].sort( - (a, b) => a.sequence - b.sequence, - ); - }); - } - } catch (error) { - console.error("[Progress] Refetch failed:", error); } - }, delayMs); + + const eventsResponse = await authenticatedFetch( + api.progress.sessionEvents(currentSession.sessionId, tenantId, currentSession.serialNumber), + getAccessToken, + ); + if (eventsResponse.ok) { + const eventsData = await eventsResponse.json(); + const fetched: EnrollmentEvent[] = eventsData.events || []; + setEvents((prev) => { + const existingIds = new Set(prev.map((e) => e.eventId)); + const newEvents = fetched.filter((e) => !existingIds.has(e.eventId)); + if (newEvents.length === 0) return prev; + return [...prev, ...newEvents].sort( + (a, b) => a.sequence - b.sequence, + ); + }); + } + } catch (error) { + console.error("[Progress] Refetch failed:", error); + } }, [tenantId, getAccessToken, setSession], ); useEffect(() => { - return () => { - if (refetchTimerRef.current) clearTimeout(refetchTimerRef.current); - }; + refreshRef.current = refresh; + }, [refresh]); + + const scheduleFetchEvents = useCallback(() => { + refreshScheduler.current ??= createBurstScheduler( + () => { void refreshRef.current(); }, + { trailingMs: REFRESH_TRAILING_MS, maxWaitMs: REFRESH_MAX_WAIT_MS }, + ); + refreshScheduler.current.trigger(); + }, []); + + // Drop a pending trailing refetch on unmount + useEffect(() => { + return () => refreshScheduler.current?.cancel(); }, []); return { diff --git a/src/Web/autopilot-monitor-web/app/progress/hooks/useProgressSignalR.ts b/src/Web/autopilot-monitor-web/app/progress/hooks/useProgressSignalR.ts index 4f31018c..92d8d5b3 100644 --- a/src/Web/autopilot-monitor-web/app/progress/hooks/useProgressSignalR.ts +++ b/src/Web/autopilot-monitor-web/app/progress/hooks/useProgressSignalR.ts @@ -28,7 +28,7 @@ interface UseProgressSignalRParams { session: Session | null; sessionRef: React.RefObject; signalR: SignalRApi; - scheduleFetchEvents: (delayMs?: number) => void; + scheduleFetchEvents: () => void; addNotification: AddNotification; } @@ -94,7 +94,7 @@ export function useProgressSignalR({ const scheduleRefetch = (source: string, sessionId: string) => { if (!sessionRef.current || sessionId !== sessionRef.current.sessionId) return; console.log(`[Progress] ${source} signal for current session, scheduling refetch`); - scheduleFetchEvents(500); + scheduleFetchEvents(); }; const handleNewEvents = (data: { sessionId: string }) => { diff --git a/src/Web/autopilot-monitor-web/app/sessions/hooks/useSessionEvents.ts b/src/Web/autopilot-monitor-web/app/sessions/hooks/useSessionEvents.ts index 2cf160b9..363b8733 100644 --- a/src/Web/autopilot-monitor-web/app/sessions/hooks/useSessionEvents.ts +++ b/src/Web/autopilot-monitor-web/app/sessions/hooks/useSessionEvents.ts @@ -3,6 +3,7 @@ import { useCallback, useEffect, useRef, useState } from "react"; import { api } from "@/lib/api"; import { authenticatedFetch, TokenExpiredError } from "@/lib/authenticatedFetch"; +import { createBurstScheduler, type BurstScheduler } from "@/lib/burstScheduler"; import { extractContinuation, MAX_EAGER_PAGES } from "@/lib/paginationLink"; import { isGuid } from "@/utils/inputValidation"; import { isTerminalStatus } from "@/utils/sessionStatus"; @@ -11,6 +12,14 @@ import type { NotificationType } from "@/contexts/NotificationContext"; const TIMELINE_PAGE_SIZE = 200; +// Live refetch coalescing on SignalR signals (see lib/burstScheduler): the first signal +// of a burst refetches at once, later ones refetch 300 ms after the burst's last signal, +// and a burst that never pauses still refetches every second. Agent uploads arrive as +// bursts of batches ~0.3 s apart with seconds of silence in between, so this keeps the +// real-time feel while a burst costs one or two page reads instead of one per batch. +const EVENT_REFRESH_TRAILING_MS = 300; +const EVENT_REFRESH_MAX_WAIT_MS = 1_000; + // Single-shot refetch delay after the session transitions to a terminal status. // EnrollmentTerminationHandler emits trailing events (enrollment_summary_shown, // app_tracking_summary, diagnostics_collecting, diagnostics_uploaded) within ~5-10s @@ -75,7 +84,8 @@ export interface UseSessionEventsReturn { events: EnrollmentEvent[]; setEvents: React.Dispatch>; fetchEvents: () => Promise; - scheduleFetchEvents: (delayMs?: number) => void; + /** Coalesced live refetch — call on every SignalR signal, the scheduler decides when to fetch. */ + scheduleFetchEvents: () => void; /** * True while a Pattern-A eager-fetch is still streaming pages after the first * batch has rendered. Surfaces a "loading more events…" indicator on the @@ -88,7 +98,7 @@ export interface UseSessionEventsReturn { * Owns the session detail page's event list lifecycle: * - fetch events from Table Storage (canonical truth) * - in-flight dedup (SignalR + 30s timer + group-join can overlap) - * - debounced scheduleFetchEvents to absorb bursts + * - burst-coalesced scheduleFetchEvents (leading + trailing + max wait) * - empty-refresh guard: ignores transient empty lists, keeps last known-good * - terminal-event detection: triggers session refetch if SignalR status delta was lost * - triggers initial fetch once sessionTenantId is known @@ -109,14 +119,15 @@ export function useSessionEvents({ const [events, setEvents] = useState([]); const [isStreamingMore, setIsStreamingMore] = useState(false); - // Debounce real-time event refreshes to avoid burst reads in Table Storage. - const eventRefreshTimeoutRef = useRef(null); // Deduplication: track in-flight fetchEvents to avoid concurrent calls const fetchEventsInFlight = useRef(false); const fetchEventsQueued = useRef(false); // The queued follow-up re-invokes fetchEvents through this ref: a memoized callback // that calls itself directly cannot be memoized by the React compiler. const fetchEventsRef = useRef<() => Promise>(async () => {}); + // Created lazily on the first signal (refs stay out of render); runs through + // fetchEventsRef so it always calls the latest fetchEvents. + const eventRefreshScheduler = useRef(null); const fetchEvents = useCallback(async () => { // Deduplication: if a fetch is already in flight, queue one follow-up instead of @@ -244,14 +255,13 @@ export function useSessionEvents({ fetchEventsRef.current = fetchEvents; }, [fetchEvents]); - const scheduleFetchEvents = useCallback((delayMs = 300) => { - if (eventRefreshTimeoutRef.current) { - clearTimeout(eventRefreshTimeoutRef.current); - } - eventRefreshTimeoutRef.current = setTimeout(() => { - fetchEvents(); - }, delayMs); - }, [fetchEvents]); + const scheduleFetchEvents = useCallback(() => { + eventRefreshScheduler.current ??= createBurstScheduler( + () => { void fetchEventsRef.current(); }, + { trailingMs: EVENT_REFRESH_TRAILING_MS, maxWaitMs: EVENT_REFRESH_MAX_WAIT_MS }, + ); + eventRefreshScheduler.current.trigger(); + }, []); // Fetch events when we have the session's tenant ID useEffect(() => { @@ -264,13 +274,9 @@ export function useSessionEvents({ // eslint-disable-next-line react-hooks/exhaustive-deps }, [sessionTenantId, sessionId]); - // Clear debounce timer on unmount + // Drop a pending trailing refetch on unmount useEffect(() => { - return () => { - if (eventRefreshTimeoutRef.current) { - clearTimeout(eventRefreshTimeoutRef.current); - } - }; + return () => eventRefreshScheduler.current?.cancel(); }, []); // One-shot trailing-events refetch on terminal transition. diff --git a/src/Web/autopilot-monitor-web/app/sessions/hooks/useSessionSignalR.ts b/src/Web/autopilot-monitor-web/app/sessions/hooks/useSessionSignalR.ts index b95e1b98..a0c6a774 100644 --- a/src/Web/autopilot-monitor-web/app/sessions/hooks/useSessionSignalR.ts +++ b/src/Web/autopilot-monitor-web/app/sessions/hooks/useSessionSignalR.ts @@ -25,7 +25,7 @@ interface UseSessionSignalRParams { sessionRef: React.RefObject; resolveEffectiveTenantId: () => string | null; signalR: SignalRApi; - scheduleFetchEvents: (delayMs?: number) => void; + scheduleFetchEvents: () => void; setSession: React.Dispatch>; setSessionTenantId: React.Dispatch>; fetchAnalysisResults: (reanalyze?: boolean) => Promise; @@ -76,7 +76,7 @@ export function useSessionSignalR({ // Re-fetch events after group join to catch any SignalR messages // that were sent before the client joined the session group. // The frontend deduplicates by eventId, so no duplicates. - scheduleFetchEvents(0); + scheduleFetchEvents(); }; joinAndCatchUp(); } @@ -109,7 +109,7 @@ export function useSessionSignalR({ // Rule-result + tenant-id side effects still run. const status = sessionRef.current?.status; if (!isTerminalStatus(status)) { - // Fetch full events from storage (single source of truth), but debounce bursts. + // Fetch full events from storage (single source of truth); the scheduler coalesces bursts. // Session updates arrive via the "newevents" message (tenant group) — no session // object in this signal to keep payloads minimal. scheduleFetchEvents(); diff --git a/src/Web/autopilot-monitor-web/lib/__tests__/burstScheduler.test.ts b/src/Web/autopilot-monitor-web/lib/__tests__/burstScheduler.test.ts new file mode 100644 index 00000000..bcf4a3a3 --- /dev/null +++ b/src/Web/autopilot-monitor-web/lib/__tests__/burstScheduler.test.ts @@ -0,0 +1,98 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import { createBurstScheduler } from "../burstScheduler"; + +/** + * The scheduler decides WHEN the session timeline refetches on SignalR signals. + * Shape under test (trailing 300 ms, max wait 1000 ms — the session page's values): + * - an isolated signal refetches at once (real-time feel), + * - a short burst refetches at its start and once after it ends, + * - a stream that never pauses refetches exactly once per second. + */ + +const TRAILING = 300; +const MAX_WAIT = 1000; + +function harness() { + const runs: number[] = []; + const t0 = Date.now(); + const scheduler = createBurstScheduler(() => runs.push(Date.now() - t0), { + trailingMs: TRAILING, + maxWaitMs: MAX_WAIT, + }); + return { runs, scheduler }; +} + +describe("createBurstScheduler", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + afterEach(() => { + vi.useRealTimers(); + }); + + it("runs an isolated trigger immediately (leading edge)", () => { + const { runs, scheduler } = harness(); + scheduler.trigger(); + expect(runs).toEqual([0]); + vi.advanceTimersByTime(5_000); + expect(runs).toEqual([0]); + }); + + it("runs a short burst at its start and once after it ends", () => { + const { runs, scheduler } = harness(); + scheduler.trigger(); // 0 ms + vi.advanceTimersByTime(200); + scheduler.trigger(); // 200 ms + vi.advanceTimersByTime(200); + scheduler.trigger(); // 400 ms + vi.advanceTimersByTime(5_000); + // 0 = leading, 700 = trailing (400 + 300); the 200-ms trigger was absorbed. + expect(runs).toEqual([0, 700]); + }); + + it("caps a stream that never pauses at one run per max wait", () => { + const { runs, scheduler } = harness(); + for (let t = 0; t <= 3_000; t += 100) { + scheduler.trigger(); + vi.advanceTimersByTime(100); + } + vi.advanceTimersByTime(5_000); + // Leading at 0, then max-wait bound at 1000/2000/3000, then the trailing run + // 300 ms after the last trigger (3000 + 300). + expect(runs).toEqual([0, 1000, 2000, 3000, 3300]); + }); + + it("does not run twice inside one trailing window right after a run", () => { + const { runs, scheduler } = harness(); + scheduler.trigger(); // leading at 0 + vi.advanceTimersByTime(100); + scheduler.trigger(); // 100 ms: cool-down, becomes a trailing run at 400 + vi.advanceTimersByTime(5_000); + expect(runs).toEqual([0, 400]); + }); + + it("is leading again once the trailing window has been quiet", () => { + const { runs, scheduler } = harness(); + scheduler.trigger(); // 0 + vi.advanceTimersByTime(TRAILING); + scheduler.trigger(); // 300: quiet for a full trailing window since the last run + expect(runs).toEqual([0, 300]); + }); + + it("cancel drops the pending trailing run only", () => { + const { runs, scheduler } = harness(); + scheduler.trigger(); // 0 + vi.advanceTimersByTime(100); + scheduler.trigger(); // pending trailing at 400 + scheduler.cancel(); + vi.advanceTimersByTime(5_000); + expect(runs).toEqual([0]); + // Cancel is not a reset: a later trigger still follows the normal rules. + scheduler.trigger(); + expect(runs).toEqual([0, 5_100]); + }); + + it("rejects a max wait shorter than the trailing window", () => { + expect(() => createBurstScheduler(() => {}, { trailingMs: 500, maxWaitMs: 100 })).toThrow(); + }); +}); diff --git a/src/Web/autopilot-monitor-web/lib/burstScheduler.ts b/src/Web/autopilot-monitor-web/lib/burstScheduler.ts new file mode 100644 index 00000000..98836a2d --- /dev/null +++ b/src/Web/autopilot-monitor-web/lib/burstScheduler.ts @@ -0,0 +1,79 @@ +/** + * Burst scheduler for real-time refetches: leading edge + trailing debounce + max wait. + * + * Telemetry arrives in bursts (several agent batches within a second, then seconds of + * silence) and every batch raises a SignalR signal. A plain trailing debounce either + * fires once per batch (window shorter than the batch spacing) or hides the whole burst + * until it pauses (window longer than the spacing, timer keeps resetting). This scheduler + * keeps the live feel and still coalesces: + * + * - leading: the first trigger after a quiet period runs immediately (0 ms delay); + * - trailing: further triggers inside a burst run once, `trailingMs` after the last one, + * so the refetch follows the END of the burst instead of a fixed tick; + * - max wait: a burst that never pauses still runs at least every `maxWaitMs` + * (spool replay after a reconnect or reboot). + * + * A steady stream of triggers spaced closer than `trailingMs` therefore runs exactly once + * per `maxWaitMs`; isolated triggers run at once. Pure timers, no React — the hooks own + * the "what to run" via a ref so the callback always sees the latest closure. + */ +export interface BurstSchedulerOptions { + /** Quiet time after the last trigger before the trailing run (ms). */ + trailingMs: number; + /** Upper bound between two runs while triggers keep arriving (ms); >= trailingMs. */ + maxWaitMs: number; +} + +export interface BurstScheduler { + /** Request a run; coalesced per the rules above. */ + trigger: () => void; + /** Drop a pending trailing run (unmount). Does not reset the leading-edge cool-down. */ + cancel: () => void; +} + +export function createBurstScheduler(run: () => void, options: BurstSchedulerOptions): BurstScheduler { + const { trailingMs, maxWaitMs } = options; + if (!(trailingMs >= 0) || !(maxWaitMs >= trailingMs)) { + throw new Error(`createBurstScheduler: need 0 <= trailingMs <= maxWaitMs, got ${trailingMs}/${maxWaitMs}`); + } + + let timer: ReturnType | null = null; + // Time of the last run; the max-wait window for the next trailing run starts here. + let lastRunAt = Number.NEGATIVE_INFINITY; + + const fire = () => { + timer = null; + lastRunAt = Date.now(); + run(); + }; + + const scheduleTrailing = (now: number) => { + if (timer) clearTimeout(timer); + const fireAt = Math.min(now + trailingMs, lastRunAt + maxWaitMs); + timer = setTimeout(fire, Math.max(0, fireAt - now)); + }; + + return { + trigger: () => { + const now = Date.now(); + if (timer) { + // Inside a burst: push the trailing run out, but never past the max-wait bound. + scheduleTrailing(now); + return; + } + if (now - lastRunAt >= trailingMs) { + // Quiet long enough: leading edge, run right away. + fire(); + return; + } + // Just ran: cool down into a trailing run so a burst cannot run twice per trailingMs. + scheduleTrailing(now); + }, + cancel: () => { + if (timer) { + clearTimeout(timer); + timer = null; + } + }, + }; +}