fix(dashboards): cancel in-flight tile stream on refresh

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 })

frontend/src/scenes/dashboard/dashboardLogic.tsx

@@ -834,6 +834,9 @@ export interface dashboardLogicActions {
     setWidgetRunResults: (results: Record<number, DashboardWidgetRunResultApi>) => {
         results: Record<number, DashboardWidgetRunResultApi>
     }
+    tileStreamCancelled: () => {
+        value: true
+    }
     tileStreamingComplete: () => {
         value: true
     }
@@ -1173,6 +1176,8 @@ export const dashboardLogic = kea<dashboardLogicType>([
         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<dashboardLogicType>([
                     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<dashboardLogicType>([
                         },
                         // 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<InsightModel>)
@@ -1407,10 +1420,16 @@ export const dashboardLogic = kea<dashboardLogicType>([
                         },
                         // 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<dashboardLogicType>([
                 loadDashboardStreaming: () => true,
                 tileStreamingComplete: () => false,
                 tileStreamingFailure: () => false,
+                tileStreamCancelled: () => false,
             },
         ],
         loadingPreview: [
@@ -1971,7 +1991,13 @@ export const dashboardLogic = kea<dashboardLogicType>([
                         ...(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<dashboardLogicType>([
                 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()