diff --git a/frontend/src/scenes/dashboard/dashboardLogic.test.ts b/frontend/src/scenes/dashboard/dashboardLogic.test.ts index ca55593bc3b4..b9bf25495c02 100644 --- a/frontend/src/scenes/dashboard/dashboardLogic.test.ts +++ b/frontend/src/scenes/dashboard/dashboardLogic.test.ts @@ -1245,6 +1245,77 @@ describe('dashboardLogic', () => { }) }) + describe('tile stream cancellation and staleness', () => { + // Regression coverage for tiles reverting to stale data: a refresh used to leave the + // initial tile stream running, so a tile it delivered after the refresh silently + // overwrote the fresh result with data computed against the old filters. + it('replaces an existing tile instead of appending a duplicate for the same tile id', async () => { + logic = dashboardLogic({ id: 5 }) + logic.mount() + await expectLogic(logic).toFinishAllListeners() + + const existingTileId = logic.values.tiles[0].id + const tileCountBefore = logic.values.tiles.length + + await expectLogic(logic, () => { + logic.actions.receiveTileFromStream({ + order: 0, + tile: { id: existingTileId, layouts: {}, color: null }, + }) + }).toFinishAllListeners() + + expect(logic.values.tiles).toHaveLength(tileCountBefore) + expect(logic.values.tiles.filter((t) => t.id === existingTileId)).toHaveLength(1) + }) + + it('cancels the in-flight tile stream on refresh so a tile it delivers afterwards is dropped', async () => { + const originalEventSource = (global as any).EventSource + ;(global as any).EventSource = (global as any).EventSource ?? class {} + + const cancelStreamSpy = jest.fn() + let capturedOnMessage: ((data: any) => void) | undefined + const streamTilesSpy = jest + .spyOn(api.dashboards, 'streamTiles') + .mockImplementation(async (_id, _params, onMessage) => { + capturedOnMessage = onMessage + return cancelStreamSpy + }) + + featureFlagLogic.mount() + featureFlagLogic.actions.setFeatureFlags([FEATURE_FLAGS.SSE_DASHBOARDS], { + [FEATURE_FLAGS.SSE_DASHBOARDS]: true, + }) + + try { + logic = dashboardLogic({ id: 5 }) + logic.mount() + await expectLogic(logic).toFinishAllListeners() + + expect(streamTilesSpy).toHaveBeenCalled() + expect(capturedOnMessage).not.toBeUndefined() + + const tileCountBefore = logic.values.tiles.length + + // Simulate the user hitting Refresh while the initial stream is still delivering tiles - + // refreshDashboardItems calls this before fetching, same as a real refresh would. + await expectLogic(logic, () => { + logic.actions.abortAnyRunningQuery() + }).toFinishAllListeners() + + expect(cancelStreamSpy).toHaveBeenCalled() + + // A tile computed before the refresh, delivered late by the cancelled stream, must not land. + capturedOnMessage?.({ type: 'tile', order: 0, tile: { id: 99999, layouts: {}, color: null } }) + + expect(logic.values.tiles).toHaveLength(tileCountBefore) + expect(logic.values.tiles.find((t) => t.id === 99999)).toBeUndefined() + } finally { + streamTilesSpy.mockRestore() + ;(global as any).EventSource = originalEventSource + } + }) + }) + describe('when a dashboard item API errors', () => { beforeEach(() => { logic = dashboardLogic({ id: 8 }) diff --git a/frontend/src/scenes/dashboard/dashboardLogic.tsx b/frontend/src/scenes/dashboard/dashboardLogic.tsx index c943d7a0fdea..7eccada492c5 100644 --- a/frontend/src/scenes/dashboard/dashboardLogic.tsx +++ b/frontend/src/scenes/dashboard/dashboardLogic.tsx @@ -834,6 +834,9 @@ export interface dashboardLogicActions { setWidgetRunResults: (results: Record) => { results: Record } + tileStreamCancelled: () => { + value: true + } tileStreamingComplete: () => { value: true } @@ -1173,6 +1176,8 @@ export const dashboardLogic = kea([ tileStreamingComplete: true, /** Tile streaming failed. */ tileStreamingFailure: (error: any) => ({ error }), + /** The in-flight tile stream was cancelled (e.g. a refresh superseded it) before it completed. */ + tileStreamCancelled: true, /** A non-404 stream failure left no dashboard to render — show a load error, not "not found". */ setDashboardStreamFailed: true, /** Expose additional information about the current dashboard load in dashboardLoadData. */ @@ -1387,8 +1392,13 @@ export const dashboardLogic = kea([ await breakpoint(200) actions.resetIntermittentFilters() + // A previous stream (e.g. from an earlier load) may still be delivering tiles - + // cancel it so its late, stale-filtered results can't land after this one's. + cache.cancelTileStream?.() + const generation = (cache.tileStreamGeneration = (cache.tileStreamGeneration ?? 0) + 1) + // Start unified streaming - metadata followed by tiles - api.dashboards.streamTiles( + cache.cancelTileStream = await api.dashboards.streamTiles( props.id, { layoutSize: values.currentLayoutSize, @@ -1397,6 +1407,9 @@ export const dashboardLogic = kea([ }, // onMessage callback - handles both metadata and tiles (data) => { + if (generation !== cache.tileStreamGeneration) { + return // Superseded by a newer load/refresh - drop the stale message + } if (data.type === 'metadata') { actions.loadDashboardMetadataSuccess( getQueryBasedDashboard(data.dashboard as DashboardType) @@ -1407,10 +1420,16 @@ export const dashboardLogic = kea([ }, // onComplete callback () => { + if (generation !== cache.tileStreamGeneration) { + return + } actions.tileStreamingComplete() }, // onError callback (error) => { + if (generation !== cache.tileStreamGeneration) { + return + } console.error('❌ Tile streaming error:', error) actions.tileStreamingFailure(error) } @@ -1733,6 +1752,7 @@ export const dashboardLogic = kea([ loadDashboardStreaming: () => true, tileStreamingComplete: () => false, tileStreamingFailure: () => false, + tileStreamCancelled: () => false, }, ], loadingPreview: [ @@ -1971,7 +1991,13 @@ export const dashboardLogic = kea([ ...(tile.insight != null ? { insight: getQueryBasedInsightModel(tile.insight) } : {}), } - let newTiles = [...state.tiles, transformedTile] + // Overlapping streams (or a stream re-delivering a tile) should replace the + // existing entry rather than append a duplicate for the same tile id. + const existingIndex = state.tiles.findIndex((t) => t.id === transformedTile.id) + const newTiles = + existingIndex >= 0 + ? state.tiles.map((t, i) => (i === existingIndex ? transformedTile : t)) + : [...state.tiles, transformedTile] return { ...state, @@ -4118,6 +4144,14 @@ export const dashboardLogic = kea([ cache.abortController.abort() cache.abortController = null } + if (cache.cancelTileStream) { + // Bump the generation first so any tile the stream already had in flight is + // dropped by the loader's guard, then tear down the connection itself. + cache.tileStreamGeneration = (cache.tileStreamGeneration ?? 0) + 1 + cache.cancelTileStream() + cache.cancelTileStream = null + actions.tileStreamCancelled() + } }, cancelDashboardRefresh: () => { actions.abortAnyRunningQuery()