This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 075c9462b96 [v3-3-test] UI: Fix task states stuck stale when a run
finishes quickly (#70319) (#70397)
075c9462b96 is described below
commit 075c9462b96bd75183d38331e1ac26da6bc99fc7
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Sat Jul 25 08:49:20 2026 +0300
[v3-3-test] UI: Fix task states stuck stale when a run finishes quickly
(#70319) (#70397)
The Grid/Graph TI-summaries stream re-fetches on an interval that is
gated on the dag run still being active. When a run reaches a terminal
state before the next tick fires — common for sub-interval runs on
low-latency deployments — the interval is cleared without ever fetching
the final task states, so node badges and grid cells stay frozen on
whatever the last stream saw (often "no status") until a full remount.
Re-stream once on the active→terminal transition so the final states
always land, mirroring the self-gating behaviour of the polled run
queries which never stop before observing a terminal state.
(cherry picked from commit 47a0324461fd1d6b2acb296e56498a3684f8407d)
Co-authored-by: Andrew Chang <[email protected]>
---
.../ui/src/queries/useGridTISummaries.test.tsx | 93 ++++++++++++++++++++++
.../airflow/ui/src/queries/useGridTISummaries.ts | 13 ++-
2 files changed, 105 insertions(+), 1 deletion(-)
diff --git
a/airflow-core/src/airflow/ui/src/queries/useGridTISummaries.test.tsx
b/airflow-core/src/airflow/ui/src/queries/useGridTISummaries.test.tsx
index f54e65efec3..c0ea5457ac5 100644
--- a/airflow-core/src/airflow/ui/src/queries/useGridTISummaries.test.tsx
+++ b/airflow-core/src/airflow/ui/src/queries/useGridTISummaries.test.tsx
@@ -26,6 +26,8 @@ import {
useGridServiceGetGridRunsKey,
useTaskInstanceServiceGetTaskInstancesKey,
} from "openapi/queries";
+import type { TaskInstanceState } from "openapi/requests";
+import { useAutoRefresh } from "src/utils";
import { useGridTiSummariesStream } from "./useGridTISummaries";
@@ -257,4 +259,95 @@ describe("useGridTiSummariesStream", () => {
// The state should NOT be updated with the aborted stream's value
expect(result.current.summariesByRunId.get("run_1")).toBeUndefined();
});
+
+ it("re-streams one final time when runs finish before the next interval
tick", async () => {
+ const queryClient = new QueryClient({
+ defaultOptions: {
+ queries: {
+ gcTime: Infinity,
+ staleTime: Infinity,
+ },
+ },
+ });
+ const wrapper = createWrapper(queryClient);
+
+ vi.mocked(useAutoRefresh).mockReturnValue(3000);
+
+ const runningChunk = `${JSON.stringify({ run_id: "run_1", state:
"running", task_id: "task_1" })}\n`;
+ const successChunk = `${JSON.stringify({ run_id: "run_1", state:
"success", task_id: "task_1" })}\n`;
+
+ mockFetch
+ .mockResolvedValueOnce(createMockResponse([runningChunk]))
+ .mockResolvedValueOnce(createMockResponse([successChunk]));
+
+ const { rerender, result } = renderHook(
+ ({ states }: { states: Array<TaskInstanceState> }) =>
+ useGridTiSummariesStream({ dagId: "dag_1", runIds: ["run_1"], states
}),
+ { initialProps: { states: ["running"] as Array<TaskInstanceState> },
wrapper },
+ );
+
+ await act(async () => {
+ await vi.advanceTimersByTimeAsync(1);
+ });
+
+ expect(mockFetch).toHaveBeenCalledTimes(1);
+ expect(result.current.summariesByRunId.get("run_1")).toEqual({
+ run_id: "run_1",
+ state: "running",
+ task_id: "task_1",
+ });
+
+ // The run reaches a terminal state before the 3s interval ever ticks
+ rerender({ states: ["success"] as Array<TaskInstanceState> });
+
+ await act(async () => {
+ await vi.advanceTimersByTimeAsync(1);
+ });
+
+ expect(mockFetch).toHaveBeenCalledTimes(2);
+ expect(result.current.summariesByRunId.get("run_1")).toEqual({
+ run_id: "run_1",
+ state: "success",
+ task_id: "task_1",
+ });
+
+ // No interval remains armed once the runs are terminal
+ await act(async () => {
+ await vi.advanceTimersByTimeAsync(10_000);
+ });
+
+ expect(mockFetch).toHaveBeenCalledTimes(2);
+ });
+
+ it("does not re-stream when runs are already terminal on mount", async () =>
{
+ const queryClient = new QueryClient({
+ defaultOptions: {
+ queries: {
+ gcTime: Infinity,
+ staleTime: Infinity,
+ },
+ },
+ });
+ const wrapper = createWrapper(queryClient);
+
+ vi.mocked(useAutoRefresh).mockReturnValue(3000);
+
+ mockFetch.mockImplementation(() =>
Promise.resolve(createMockResponse([])));
+
+ renderHook(
+ () =>
+ useGridTiSummariesStream({
+ dagId: "dag_1",
+ runIds: ["run_1"],
+ states: ["success"] as Array<TaskInstanceState>,
+ }),
+ { wrapper },
+ );
+
+ await act(async () => {
+ await vi.advanceTimersByTimeAsync(10_000);
+ });
+
+ expect(mockFetch).toHaveBeenCalledTimes(1);
+ });
});
diff --git a/airflow-core/src/airflow/ui/src/queries/useGridTISummaries.ts
b/airflow-core/src/airflow/ui/src/queries/useGridTISummaries.ts
index d90b2f0078e..eab4485c86d 100644
--- a/airflow-core/src/airflow/ui/src/queries/useGridTISummaries.ts
+++ b/airflow-core/src/airflow/ui/src/queries/useGridTISummaries.ts
@@ -17,7 +17,7 @@
* under the License.
*/
import { useQueryClient } from "@tanstack/react-query";
-import { useEffect, useState } from "react";
+import { useEffect, useRef, useState } from "react";
import {
useDagRunServiceGetDagRunsKey,
@@ -155,6 +155,17 @@ export const useGridTiSummariesStream = ({
return () => clearInterval(timer);
}, [hasActiveRuns, baseRefetchInterval]);
+ // The interval above is cleared the moment the last active run turns
terminal, which can happen
+ // before its first tick for fast runs — re-stream once on that transition
so final TI states land.
+ const wasActiveRef = useRef(false);
+
+ useEffect(() => {
+ if (wasActiveRef.current && !hasActiveRuns) {
+ setRefreshTick((tick) => tick + 1);
+ }
+ wasActiveRef.current = hasActiveRuns;
+ }, [hasActiveRuns]);
+
// Re-stream whenever a mutation invalidates a grid-related query (TI states,
// run states, or grid structure). Invalidation events only fire from
explicit
// invalidateQueries() calls — never from polling intervals — so this never
