fix(ui): stop a superseded Test Connection run from clobbering a newer result

.github/playwright/impact-map.json

@@ -446,6 +446,34 @@
         "playwright/e2e/Features/MutuallyExclusiveColumnTags.spec.ts",
         "playwright/e2e/Features/ColumnBulkOperationsTagsGlossary.spec.ts"
       ]
+    },
+    {
+      "sources": [
+        "openmetadata-ui/src/main/resources/ui/src/components/common/TestConnection/**",
+        "openmetadata-ui/src/main/resources/ui/src/utils/TestConnectionModalUtils.tsx"
+      ],
+      "projects": ["Basic"],
+      "specs": ["playwright/e2e/Flow/TestConnectionModal.spec.ts"]
+    },
+    {
+      "sources": [
+        "openmetadata-ui/src/main/resources/ui/src/components/common/TestConnection/**",
+        "openmetadata-ui/src/main/resources/ui/src/utils/TestConnectionModalUtils.tsx"
+      ],
+      "projects": ["chromium"],
+      "specs": ["playwright/e2e/Flow/ConnectionConfigLayout.spec.ts"]
+    },
+    {
+      "sources": [
+        "openmetadata-ui/src/main/resources/ui/src/components/common/TestConnection/**",
+        "openmetadata-ui/src/main/resources/ui/src/utils/TestConnectionModalUtils.tsx"
+      ],
+      "projects": ["Ingestion"],
+      "specs": [
+        "playwright/e2e/Flow/ServiceForm.spec.ts",
+        "playwright/e2e/Flow/ApiServiceRest.spec.ts",
+        "playwright/e2e/Features/AutoPilot.spec.ts"
+      ]
     }
   ]
 }

openmetadata-ui/src/main/resources/ui/src/components/common/TestConnection/TestConnection.test.tsx

@@ -1176,4 +1176,157 @@ describe('Test Connection Component', () => {
 
     expect(onTestConnectionStatusChange).toHaveBeenLastCalledWith(true);
   });
+
+  it('Should not let a superseded run overwrite the result of a newer one', async () => {
+    jest.useFakeTimers();
+
+    const onTestConnectionStatusChange = jest.fn();
+
+    // the first run dies while polling, which leaves its expiry timer armed
+    (getWorkflowById as jest.Mock).mockRejectedValueOnce(new Error('failed'));
+
+    await act(async () => {
+      render(
+        <TestConnection
+          {...mockProps}
+          onTestConnectionStatusChange={onTestConnectionStatusChange}
+        />
+      );
+    });
+
+    await act(async () => {
+      fireEvent.click(screen.getByTestId('test-connection-btn'));
+    });
+
+    await act(async () => {
+      jest.advanceTimersByTime(2000);
+    });
+
+    expect(
+      screen.getByText('message.connection-test-failed')
+    ).toBeInTheDocument();
+
+    await act(async () => {
+      fireEvent.click(screen.getByTestId('test-connection-btn'));
+    });
+
+    await act(async () => {
+      jest.advanceTimersByTime(2000);
+    });
+
+    expect(onTestConnectionStatusChange).toHaveBeenLastCalledWith(true);
+
+    // past the first run's expiry, whose callback must no longer touch the UI
+    await act(async () => {
+      jest.advanceTimersByTime(180000);
+    });
+
+    expect(onTestConnectionStatusChange).toHaveBeenLastCalledWith(true);
+    expect(
+      screen.queryByTestId('connection-timeout-message')
+    ).not.toBeInTheDocument();
+  });
+
+  it('Should ignore a superseded run whose poll resolves after a newer run started', async () => {
+    jest.useFakeTimers();
+
+    let resolveStalePoll: (value: unknown) => void = () => undefined;
+    const stalePoll = new Promise((resolve) => {
+      resolveStalePoll = resolve;
+    });
+
+    (getWorkflowById as jest.Mock)
+      .mockImplementationOnce(() => stalePoll)
+      .mockImplementation(() =>
+        Promise.resolve({
+          ...WORKFLOW_DETAILS,
+          status: 'Running',
+          response: { ...WORKFLOW_DETAILS.response, status: 'Running' },
+        })
+      );
+
+    await act(async () => {
+      render(<TestConnection {...mockProps} />);
+    });
+
+    await act(async () => {
+      fireEvent.click(screen.getByTestId('test-connection-btn'));
+    });
+
+    // the first run's poll is now in flight and will not settle yet
+    await act(async () => {
+      jest.advanceTimersByTime(2000);
+    });
+
+    // its expiry re-enables the button while that request is still pending
+    await act(async () => {
+      jest.advanceTimersByTime(180000);
+    });
+
+    await act(async () => {
+      fireEvent.click(screen.getByTestId('test-connection-btn'));
+    });
+
+    await act(async () => {
+      jest.advanceTimersByTime(2000);
+    });
+
+    const callsBeforeStaleResolves = (getWorkflowById as jest.Mock).mock.calls
+      .length;
+
+    await act(async () => {
+      resolveStalePoll({
+        ...WORKFLOW_DETAILS,
+        status: 'Failed',
+        response: { ...WORKFLOW_DETAILS.response, status: 'Failed' },
+      });
+    });
+
+    expect(
+      screen.queryByText('message.connection-test-failed')
+    ).not.toBeInTheDocument();
+
+    // the newer run must still own its timers and keep polling
+    await act(async () => {
+      jest.advanceTimersByTime(2000);
+    });
+
+    expect((getWorkflowById as jest.Mock).mock.calls.length).toBeGreaterThan(
+      callsBeforeStaleResolves
+    );
+  });
+
+  it('Should stop polling the workflow after unmount', async () => {
+    jest.useFakeTimers();
+
+    (getWorkflowById as jest.Mock).mockImplementation(() =>
+      Promise.resolve({
+        ...WORKFLOW_DETAILS,
+        status: 'Running',
+        response: { ...WORKFLOW_DETAILS.response, status: 'Running' },
+      })
+    );
+
+    const { unmount } = render(<TestConnection {...mockProps} />);
+
+    await act(async () => {
+      fireEvent.click(screen.getByTestId('test-connection-btn'));
+    });
+
+    await act(async () => {
+      jest.advanceTimersByTime(2000);
+    });
+
+    unmount();
+
+    const callsAtUnmount = (getWorkflowById as jest.Mock).mock.calls.length;
+
+    await act(async () => {
+      jest.advanceTimersByTime(180000);
+    });
+
+    expect((getWorkflowById as jest.Mock).mock.calls).toHaveLength(
+      callsAtUnmount
+    );
+  });
 });

openmetadata-ui/src/main/resources/ui/src/components/common/TestConnection/TestConnection.tsx

@@ -168,6 +168,17 @@ const TestConnection: FC<TestConnectionProps> = ({
    */
   const currentWorkflowRef = useRef(currentWorkflow);
 
+  // Timers live in a ref so a new run - or an unmount - can cancel the previous
+  // run's callbacks before they write state over a newer result.
+  const timersRef = useRef<{ intervalId?: number; timeoutId?: number }>({});
+  const runIdRef = useRef(0);
+
+  const clearTimers = useCallback(() => {
+    clearInterval(timersRef.current.intervalId);
+    clearTimeout(timersRef.current.timeoutId);
+    timersRef.current = {};
+  }, []);
+
   const { controller } = useAbortController();
 
   const serviceType = useMemo(() => {
@@ -326,25 +337,36 @@ const TestConnection: FC<TestConnectionProps> = ({
   const handleWorkflowPolling = async (
     response: Workflow,
     definitionSteps: TestConnectionStep[],
-    intervalObject: {
-      intervalId?: number;
-      timeoutId?: number;
-    }
+    runId: number
   ) => {
     // return a promise that wraps the interval and handles errors inside it
     return new Promise<void>((resolve, reject) => {
       /**
        * fetch workflow repeatedly with 2s interval
        * until status is either Failed or Successful
        */
-      intervalObject.intervalId = toNumber(
+      timersRef.current.intervalId = toNumber(
         setInterval(async () => {
+          if (runId !== runIdRef.current) {
+            resolve();
+
+            return;
+          }
+
           setProgress(updateProgress);
           try {
             const workflowResponse = await getWorkflowData(
               response.id,
               controller.signal
             );
+
+            // a newer run may have started while the request was in flight
+            if (runId !== runIdRef.current) {
+              resolve();
+
+              return;
+            }
+
             const { response: testConnectionResponse } = workflowResponse;
             const { status: testConnectionStatus, steps = [] } =
               testConnectionResponse || {};
@@ -367,9 +389,7 @@ const TestConnection: FC<TestConnectionProps> = ({
               definitionSteps
             );
 
-            // clear the current interval
-            clearInterval(intervalObject.intervalId);
-            clearTimeout(intervalObject.timeoutId);
+            clearTimers();
 
             // set testing connection to false
             setIsTestingConnection(false);
@@ -393,13 +413,10 @@ const TestConnection: FC<TestConnectionProps> = ({
     setMessage(t(TEST_CONNECTION_TESTING_MESSAGE));
     handleResetState();
 
-    const updatedFormData = formatFormDataForSubmit(getData());
+    clearTimers();
+    const runId = ++runIdRef.current;
 
-    // current interval id
-    const intervalObject: {
-      intervalId?: number;
-      timeoutId?: number;
-    } = {};
+    const updatedFormData = formatFormDataForSubmit(getData());
 
     const { ingestionRunner, ...rest } = updatedFormData as ConfigObject & {
       ingestionRunner?: string;
@@ -453,8 +470,12 @@ const TestConnection: FC<TestConnectionProps> = ({
 
       // stop fetching the workflow after 2 minutes
       const timeoutId = setTimeout(() => {
+        if (runId !== runIdRef.current) {
+          return;
+        }
+
         // clear the current interval
-        clearInterval(intervalObject.intervalId);
+        clearInterval(timersRef.current.intervalId);
 
         // using reference to ensure call back should have latest value
         const currentWorkflowStatus = currentWorkflowRef.current
@@ -482,13 +503,13 @@ const TestConnection: FC<TestConnectionProps> = ({
         onTestConnectionStatusChange?.(false);
       }, FETCHING_EXPIRY_TIME);
 
-      intervalObject.timeoutId = Number(timeoutId);
+      timersRef.current.timeoutId = Number(timeoutId);
 
       // Handle workflow polling and completion
-      await handleWorkflowPolling(response, definitionSteps, intervalObject);
+      await handleWorkflowPolling(response, definitionSteps, runId);
     } catch (error) {
       setProgress(TEST_CONNECTION_PROGRESS_PERCENTAGE.HUNDRED);
-      clearInterval(intervalObject.intervalId);
+      clearTimers();
       setIsTestingConnection(false);
       setMessage(t(TEST_CONNECTION_FAILURE_MESSAGE));
       setTestStatus(StatusType.Failed);
@@ -678,6 +699,8 @@ const TestConnection: FC<TestConnectionProps> = ({
 
   useEffect(() => {
     return () => {
+      clearTimers();
+
       /**
        * if workflow is present then delete the workflow when component unmount
        */