Update polling and timeout message handling | CTT-188 (#335)

* fix(pollers): stop polling if an explicit error code is available on the member

* fix(polling): move distinctUntilChanged behavior that blocked correct handling of polls

* fix(pollcount): deprecate pollingCount in favor of real timers

* fix(nodda): wesockets detect CONNECTED before IMPAIRED is set

This change allows a proper detection of NO DDA flows in verification mode even when websockets are insanely fast in their detection

* refactor: apply some code review suggestions

* Revert "fix(nodda): wesockets detect CONNECTED before IMPAIRED is set"

This reverts commit e3b16a8.

Author: codingLogan

src/hooks/__tests__/usePollMember-test.tsx

@@ -88,7 +88,6 @@ describe('usePollMember', () => {
     expect(apiValue.loadJob).toHaveBeenCalledWith(connectedMember.most_recent_job_guid)
     expect(states[0]).toMatchObject({
       isError: false,
-      pollingCount: 1,
       currentResponse: {
         member: connectedMember,
         job: JOB_DATA,
@@ -128,7 +127,6 @@ describe('usePollMember', () => {
 
     expect(states[0]).toMatchObject({
       isError: true,
-      pollingCount: 1,
       pollingIsDone: false,
     })
 
@@ -314,7 +312,7 @@ describe('usePollMember', () => {
     subscription.unsubscribe()
   })
 
-  it('should increment pollingCount on each poll', async () => {
+  it('should emit sequential states on each poll', async () => {
     const member1 = { ...member.member, guid: 'MBR-1', most_recent_job_guid: 'JOB-1' }
     const member2 = { ...member.member, guid: 'MBR-2', most_recent_job_guid: 'JOB-2' }
 
@@ -349,8 +347,14 @@ describe('usePollMember', () => {
       { timeout: 3500 },
     )
 
-    expect(states[0].pollingCount).toBe(1)
-    expect(states[1].pollingCount).toBe(2)
+    expect(states[0].currentResponse).toEqual({
+      member: member1,
+      job: { ...JOB_DATA, guid: 'JOB-1' },
+    })
+    expect(states[1].currentResponse).toEqual({
+      member: member2,
+      job: { ...JOB_DATA, guid: 'JOB-2' },
+    })
 
     subscription.unsubscribe()
   }, 10000)
@@ -871,4 +875,58 @@ describe('usePollMember', () => {
 
     subscription.unsubscribe()
   })
+
+  it('should eventually stop polling for an oauth member in an error state that was never aggregating', async () => {
+    /**
+     * This tests a specific bug where distinctUntilChanged in the transport layer
+     * blocks the second identical poll from reaching the scan accumulator.
+     *
+     * handlePollingResponse uses isNotAggregatingAtAll:
+     *   previousMember.is_being_aggregated === false && polledMember.is_being_aggregated === false
+     *
+     * Without the fix, the second poll is blocked, previousResponse stays as {} (DEFAULT),
+     * isNotAggregatingAtAll is never true, and the OAuth member polls forever.
+     */
+    const oauthMember = {
+      ...member.member,
+      connection_status: ReadableStatuses.PREVENTED,
+      is_being_aggregated: false,
+      is_oauth: true,
+    }
+
+    const apiValue = {
+      loadMemberByGuid: vi.fn().mockResolvedValue(oauthMember),
+      loadJob: vi.fn().mockResolvedValue(JOB_DATA),
+    }
+
+    const preloadedState = {
+      experimentalFeatures: {
+        memberPollingMilliseconds: 100,
+      },
+    }
+
+    const { result } = renderHook(() => usePollMember(), {
+      wrapper: createWrapper(apiValue, preloadedState),
+    })
+
+    const pollMember = result.current
+    const states: PollingState[] = []
+
+    const subscription = pollMember('MBR-123').subscribe((state: PollingState) => {
+      states.push(state)
+    })
+
+    await waitFor(
+      () => {
+        expect(states.some((s) => s.pollingIsDone === true)).toBe(true)
+      },
+      { timeout: 1000 },
+    )
+
+    expect(states.find((s) => s.pollingIsDone === true)?.userMessage).toBe(
+      CONNECTING_MESSAGES.ERROR,
+    )
+
+    subscription.unsubscribe()
+  })
 })

src/hooks/usePollMember.tsx

@@ -5,15 +5,15 @@ import { useWebSocket } from 'src/context/WebSocketContext'
 import { useSelector } from 'react-redux'
 import { getExperimentalFeatures } from 'src/redux/reducers/experimentalFeaturesSlice'
 
-import { scan } from 'rxjs/operators'
+import { scan, distinctUntilChanged } from 'rxjs/operators'
+import _isEqual from 'lodash/isEqual'
 import {
   createMemberUpdateTransport,
   MemberUpdate,
 } from 'src/utilities/transport/MemberUpdateTransport'
 
 export interface PollingState {
   isError: boolean
-  pollingCount: number
   currentResponse?: MemberUpdate | Record<string, never>
   previousResponse?: MemberUpdate | Record<string, never>
   pollingIsDone: boolean
@@ -61,8 +61,6 @@ export function usePollMember() {
           const pollingState: PollingState = {
             // only track if the most recent poll was an error
             isError,
-            // always increase polling count
-            pollingCount: acc.pollingCount + 1,
             // dont update previous response if this is an error
             previousResponse: isError ? acc.previousResponse : acc.currentResponse,
             // dont update current response if this is an error
@@ -93,6 +91,35 @@ export function usePollMember() {
         },
         { ...DEFAULT_POLLING_STATE } as PollingState,
       ),
+      // Deduplicate consecutive identical polling states to prevent unnecessary re-renders.
+      // This must live here — after the scan — so the scan always sees every update
+      // and can correctly track previousResponse/currentResponse transitions. Placing
+      // distinctUntilChanged earlier (in the transport) caused the scan to miss
+      // identical consecutive polls, breaking isNotAggregatingAtAll detection in
+      // handlePollingResponse and causing OAuth members to poll indefinitely.
+      distinctUntilChanged((prev: PollingState, curr: PollingState) => {
+        if (prev.isError !== curr.isError) return false
+        if (prev.pollingIsDone !== curr.pollingIsDone) return false
+        if (prev.userMessage !== curr.userMessage) return false
+        if (prev.initialDataReady !== curr.initialDataReady) return false
+
+        const prevMember = (prev.currentResponse as MemberUpdate)?.member
+        const currMember = (curr.currentResponse as MemberUpdate)?.member
+        const prevJob = (prev.currentResponse as MemberUpdate)?.job
+        const currJob = (curr.currentResponse as MemberUpdate)?.job
+
+        // Return true to *prevent* emitting the event
+        // Return false to emit the event
+        return (
+          prevMember?.connection_status === currMember?.connection_status &&
+          _isEqual(prevMember?.mfa, currMember?.mfa) &&
+          prevMember?.is_being_aggregated === currMember?.is_being_aggregated &&
+          prevMember?.most_recent_job_detail_code === currMember?.most_recent_job_detail_code &&
+          prevMember?.error?.error_code === currMember?.error?.error_code &&
+          prevJob?.guid === currJob?.guid &&
+          prevJob?.async_account_data_ready === currJob?.async_account_data_ready
+        )
+      }),
     )
   }
 

src/utilities/__tests__/pollers-test.js

@@ -5,6 +5,12 @@ import {
 } from 'src/utilities/pollers'
 import { ErrorStatuses, ProcessingStatuses, ReadableStatuses } from 'src/const/Statuses'
 
+// CHALLENGED is intentionally excluded from error assertions because
+// handlePollingResponse routes it through the MFA-specific branch and message.
+const nonChallengedErrorStatuses = ErrorStatuses.filter(
+  (status) => status !== ReadableStatuses.CHALLENGED,
+)
+
 describe('handlePollingResponse', () => {
   test('it should stop polling and update the message', () => {
     testStatus(ReadableStatuses.CHALLENGED, true, CONNECTING_MESSAGES.MFA)
@@ -62,16 +68,13 @@ describe('handlePollingResponse', () => {
 
   describe('Error states', () => {
     it('should stop polling and show a message', () => {
-      ErrorStatuses.forEach((status) => {
-        // CHALLENGED state is an error state, but has specific logic
-        if (status !== ReadableStatuses.CHALLENGED) {
-          testStatus(status, true, CONNECTING_MESSAGES.ERROR)
-        }
+      nonChallengedErrorStatuses.forEach((status) => {
+        testStatus(status, true, CONNECTING_MESSAGES.ERROR)
       })
     })
 
     it('should wait for aggregation to be done for error states', () => {
-      ErrorStatuses.forEach((status) => {
+      nonChallengedErrorStatuses.forEach((status) => {
         const pollingState = {
           ...DEFAULT_POLLING_STATE,
           currentResponse: {
@@ -82,13 +85,10 @@ describe('handlePollingResponse', () => {
           },
         }
 
-        // CHALLENGED state is an error state, but has specific logic
-        if (status !== ReadableStatuses.CHALLENGED) {
-          const [stopPolling, message] = handlePollingResponse(pollingState)
+        const [stopPolling, message] = handlePollingResponse(pollingState)
 
-          expect(stopPolling).toEqual(false)
-          expect(message).toEqual(CONNECTING_MESSAGES.VERIFYING)
-        }
+        expect(stopPolling).toEqual(false)
+        expect(message).toEqual(CONNECTING_MESSAGES.VERIFYING)
       })
     })
 
@@ -118,9 +118,9 @@ describe('handlePollingResponse', () => {
     })
   })
 
-  describe('OAuth status', () => {
+  describe('Terminal error code handling', () => {
     it('should keep polling and show the OAuth message if in error, but not finished agging', () => {
-      ErrorStatuses.forEach((status) => {
+      nonChallengedErrorStatuses.forEach((status) => {
         const pollingState = {
           ...DEFAULT_POLLING_STATE,
           currentResponse: {
@@ -132,17 +132,15 @@ describe('handlePollingResponse', () => {
           },
         }
 
-        if (status !== ReadableStatuses.CHALLENGED) {
-          const [stopPolling, message] = handlePollingResponse(pollingState)
+        const [stopPolling, message] = handlePollingResponse(pollingState)
 
-          expect(message).toEqual(CONNECTING_MESSAGES.OAUTH)
-          expect(stopPolling).toEqual(false)
-        }
+        expect(message).toEqual(CONNECTING_MESSAGES.OAUTH)
+        expect(stopPolling).toEqual(false)
       })
     })
 
     it('should go to error view if we are done aggregating', () => {
-      ErrorStatuses.forEach((status) => {
+      nonChallengedErrorStatuses.forEach((status) => {
         const pollingState = {
           ...DEFAULT_POLLING_STATE,
           currentResponse: {
@@ -161,12 +159,31 @@ describe('handlePollingResponse', () => {
           },
         }
 
-        if (status !== ReadableStatuses.CHALLENGED) {
-          const [stopPolling, message] = handlePollingResponse(pollingState)
+        const [stopPolling, message] = handlePollingResponse(pollingState)
 
-          expect(message).toEqual(CONNECTING_MESSAGES.ERROR)
-          expect(stopPolling).toEqual(true)
+        expect(message).toEqual(CONNECTING_MESSAGES.ERROR)
+        expect(stopPolling).toEqual(true)
+      })
+    })
+
+    it('should stop polling when a terminal error code is present, oauth is true, and there is no previous response', () => {
+      nonChallengedErrorStatuses.forEach((status) => {
+        const pollingState = {
+          ...DEFAULT_POLLING_STATE,
+          currentResponse: {
+            member: {
+              connection_status: status,
+              is_being_aggregated: false,
+              is_oauth: true,
+              error: { error_code: 'ANY_ERROR_CODE' },
+            },
+          },
         }
+
+        const [stopPolling, message] = handlePollingResponse(pollingState)
+
+        expect(message).toEqual(CONNECTING_MESSAGES.ERROR)
+        expect(stopPolling).toEqual(true)
       })
     })
   })

src/utilities/pollers.js

@@ -18,7 +18,6 @@ export const CONNECTING_MESSAGES = {
 
 export const DEFAULT_POLLING_STATE = {
   isError: false, // whether or not the last poll was an error
-  pollingCount: 0, // used to count how many times we have polled
   previousResponse: {}, // previous response from last poll
   currentResponse: {}, // current response
   pollingIsDone: false, // whether or not we should stop polling
@@ -65,11 +64,22 @@ export function handlePollingResponse(pollingState) {
     return [false, CONNECTING_MESSAGES.VERIFYING]
   }
 
-  // if we aren't aggregating whatsoever and in an error state, stop polling
+  // if we aren't aggregating whatsoever and in an error state, stop polling,
+  // even if we don't have an explicit error code.
   if (isNotAggregatingAtAll && ErrorStatuses.includes(polledMember.connection_status)) {
     return [true, CONNECTING_MESSAGES.ERROR]
   }
 
+  // if we aren't aggregating, are in an error state, and have an explicit error code already,
+  // stop polling and show the error message.
+  if (
+    polledMember.is_being_aggregated === false &&
+    ErrorStatuses.includes(polledMember.connection_status) &&
+    Boolean(polledMember.error?.error_code)
+  ) {
+    return [true, CONNECTING_MESSAGES.ERROR]
+  }
+
   /**
    * If this is an OAuth member, we could be stuck 'connecting' forever if the
    * user bails out of the authentication process, leaving the member in the
@@ -112,8 +122,6 @@ export function pollOauthState(oauthStateGuid, api) {
         return {
           // only track if the most recent poll was an error
           isError,
-          // always increase polling count
-          pollingCount: acc.pollingCount + 1,
           // dont update previous response if this is an error
           previousResponse: isError ? acc.previousResponse : acc.currentResponse,
           // dont update current response if this is an error

src/utilities/transport/MemberUpdateTransport.ts

@@ -1,14 +1,5 @@
 import { Observable, defer, interval, of, merge } from 'rxjs'
-import {
-  catchError,
-  map,
-  mergeMap,
-  exhaustMap,
-  filter,
-  distinctUntilChanged,
-  scan,
-} from 'rxjs/operators'
-import _isEqual from 'lodash/isEqual'
+import { catchError, map, mergeMap, exhaustMap, filter, scan } from 'rxjs/operators'
 import type { ApiContextTypes } from 'src/context/ApiContext'
 import { WebSocketConnection } from 'src/context/WebSocketContext'
 
@@ -83,26 +74,5 @@ export function createMemberUpdateTransport(
     transport$ = merge(polling$, socket$)
   }
 
-  return transport$.pipe(
-    distinctUntilChanged((prev, curr) => {
-      // Don't deduplicate errors
-      if (prev instanceof Error || curr instanceof Error) return false
-
-      const prevMember = prev.member
-      const currMember = curr.member
-
-      // Compare the relevant fields to determine if we should emit an update
-      // Return true to *prevent* emitting the event
-      // Return false to emit the event
-      return (
-        prevMember?.connection_status === currMember?.connection_status &&
-        _isEqual(prevMember?.mfa, currMember?.mfa) &&
-        prev.job?.guid === curr.job?.guid &&
-        prev.job?.async_account_data_ready === curr.job?.async_account_data_ready &&
-        prevMember?.is_being_aggregated === currMember?.is_being_aggregated &&
-        prevMember?.most_recent_job_detail_code === currMember?.most_recent_job_detail_code &&
-        prevMember?.error?.error_code === currMember?.error?.error_code
-      )
-    }),
-  )
+  return transport$
 }