From f5a5d01e55da0150ee586e17cc34f6b0853c33a2 Mon Sep 17 00:00:00 2001 From: Craig Tingey Date: Fri, 11 Sep 2026 10:03:11 -0600 Subject: [PATCH 1/4] fix: JobScheduler and Connecting know to expect a job started from firefly --- src/redux/reducers/Connect.js | 9 +- src/redux/reducers/__tests__/Connect-test.js | 25 ++ src/utilities/JobSchedule.js | 39 ++- src/utilities/__tests__/JobSchedule-test.js | 68 +++++ src/views/connecting/Connecting.js | 157 ++++++---- .../__tests__/ConnectingOAuthJobs-test.tsx | 277 ++++++++++++++++++ 6 files changed, 497 insertions(+), 78 deletions(-) create mode 100644 src/views/connecting/__tests__/ConnectingOAuthJobs-test.tsx diff --git a/src/redux/reducers/Connect.js b/src/redux/reducers/Connect.js index 336ce9dd24..ed39bdf9d3 100644 --- a/src/redux/reducers/Connect.js +++ b/src/redux/reducers/Connect.js @@ -422,7 +422,14 @@ const initializeJobSchedule = (state, action) => { const jobSchedule = JobSchedule.initialize(member, job, config, isComboJobsEnabled) - return { ...state, jobSchedule } + const members = member?.guid ? upsertMember(state, { payload: member }) : state.members + + return { + ...state, + currentMemberGuid: member?.guid ?? state.currentMemberGuid, + jobSchedule, + members, + } } const verifyExistingConnection = (state, action) => { diff --git a/src/redux/reducers/__tests__/Connect-test.js b/src/redux/reducers/__tests__/Connect-test.js index c6c467a9ef..9dc3cc5eff 100644 --- a/src/redux/reducers/__tests__/Connect-test.js +++ b/src/redux/reducers/__tests__/Connect-test.js @@ -920,6 +920,31 @@ describe('Connect redux store', () => { }, ]) }) + + test('stores the (refreshed) member as the current member without changing location', () => { + const staleMember = { guid: 'MBR-1', is_being_aggregated: false, most_recent_job_guid: null } + const freshMember = { + guid: 'MBR-1', + is_being_aggregated: true, + most_recent_job_guid: 'JOB-1', + } + const beforeState = { + ...defaultState, + currentMemberGuid: 'MBR-1', + members: [staleMember], + location: [{ step: STEPS.SEARCH }, { step: STEPS.CONNECTING }], + } + + const afterState = reducer( + beforeState, + initializeJobSchedule(freshMember, aggJob, { mode: AGG_MODE }), + ) + + expect(afterState.currentMemberGuid).toBe('MBR-1') + expect(afterState.members).toEqual([freshMember]) + expect(afterState.location).toEqual(beforeState.location) + expect(afterState.jobSchedule.isInitialized).toBe(true) + }) }) describe('RETRY_OAUTH action', () => { diff --git a/src/utilities/JobSchedule.js b/src/utilities/JobSchedule.js index 37b979a623..1ddc21187c 100644 --- a/src/utilities/JobSchedule.js +++ b/src/utilities/JobSchedule.js @@ -69,30 +69,37 @@ export const initialize = (member, recentJob, config, isComboJobsEnabled) => { /** * Update the schedule with the finished job. * - Mark the finished job as DONE - * - Find and update the next PENDING JOB + * - If nothing is left ACTIVE, promote the next PENDING job + * + * The finished job is not always the ACTIVE one. Firefly starts a job of its + * own when an OAuth member is redirected back and background aggregation is disabled, + * and that job can be the one that finishes while our scheduled job is still waiting + * to run. In that case the ACTIVE job must stay ACTIVE so Connecting can start it; + * promoting a PENDING job as well would leave two ACTIVE jobs and nothing would ever + * pick up the second one. * * @param {Object} schedule the jobSchedule object * @param {Object} finishedJob the job that was just finished * @return {Object} an updated jobSchedule */ export const onJobFinished = (schedule, finishedJob) => { - let hasSetActiveJob = false - - const updatedJobs = schedule.jobs.map((scheduledJob) => { - if (finishedJob.job_type === scheduledJob.type) { - // If the finished job's type matched the scheduled one, mark it as done - return { ...scheduledJob, status: JOB_STATUSES.DONE } - } else if (!hasSetActiveJob && scheduledJob.status === JOB_STATUSES.PENDING) { - // If we haven't set an active job and this one is pending, mark it as - // active, we only have one active job at a time. - hasSetActiveJob = true - return { ...scheduledJob, status: JOB_STATUSES.ACTIVE } - } + const jobs = schedule.jobs.map((scheduledJob) => + finishedJob?.job_type === scheduledJob.type + ? { ...scheduledJob, status: JOB_STATUSES.DONE } + : scheduledJob, + ) - return scheduledJob - }) + const hasActiveJob = jobs.some((job) => job.status === JOB_STATUSES.ACTIVE) - return { isInitialized: true, jobs: updatedJobs } + if (!hasActiveJob) { + const nextPendingIndex = jobs.findIndex((job) => job.status === JOB_STATUSES.PENDING) + + if (nextPendingIndex !== -1) { + jobs[nextPendingIndex] = { ...jobs[nextPendingIndex], status: JOB_STATUSES.ACTIVE } + } + } + + return { isInitialized: true, jobs } } export const areAllJobsDone = (schedule) => { diff --git a/src/utilities/__tests__/JobSchedule-test.js b/src/utilities/__tests__/JobSchedule-test.js index 884ac2f76b..8ca487f11c 100644 --- a/src/utilities/__tests__/JobSchedule-test.js +++ b/src/utilities/__tests__/JobSchedule-test.js @@ -219,4 +219,72 @@ describe('JobSchedule.onJobFinished', () => { }, ]) }) + + describe('when the finished job is not the active job', () => { + const verifyJob = { guid: 'JOB-2', job_type: JOB_TYPES.VERIFICATION } + + test('keeps the active job active and does not promote a pending job', () => { + const prevSchedule = { + isInitialized: true, + jobs: [ + { type: JOB_TYPES.VERIFICATION, status: JOB_STATUSES.ACTIVE }, + { type: JOB_TYPES.IDENTIFICATION, status: JOB_STATUSES.PENDING }, + ], + } + + const schedule = JobSchedule.onJobFinished(prevSchedule, aggJob) + + expect(schedule.jobs).toEqual([ + { type: JOB_TYPES.VERIFICATION, status: JOB_STATUSES.ACTIVE }, + { type: JOB_TYPES.IDENTIFICATION, status: JOB_STATUSES.PENDING }, + ]) + expect(JobSchedule.getActiveJob(schedule)).toEqual({ + type: JOB_TYPES.VERIFICATION, + status: JOB_STATUSES.ACTIVE, + }) + }) + + test('leaves the schedule alone when the finished job was already done', () => { + const prevSchedule = { + isInitialized: true, + jobs: [ + { type: JOB_TYPES.VERIFICATION, status: JOB_STATUSES.DONE }, + { type: JOB_TYPES.IDENTIFICATION, status: JOB_STATUSES.ACTIVE }, + ], + } + + const schedule = JobSchedule.onJobFinished(prevSchedule, verifyJob) + + expect(schedule.jobs).toEqual(prevSchedule.jobs) + expect(JobSchedule.areAllJobsDone(schedule)).toBe(false) + }) + + test('marks a pending job done if that is what finished, keeping the active one', () => { + const prevSchedule = { + isInitialized: true, + jobs: [ + { type: JOB_TYPES.AGGREGATION, status: JOB_STATUSES.ACTIVE }, + { type: JOB_TYPES.VERIFICATION, status: JOB_STATUSES.PENDING }, + ], + } + + const schedule = JobSchedule.onJobFinished(prevSchedule, verifyJob) + + expect(schedule.jobs).toEqual([ + { type: JOB_TYPES.AGGREGATION, status: JOB_STATUSES.ACTIVE }, + { type: JOB_TYPES.VERIFICATION, status: JOB_STATUSES.DONE }, + ]) + }) + + test('tolerates a missing job', () => { + const prevSchedule = { + isInitialized: true, + jobs: [{ type: JOB_TYPES.VERIFICATION, status: JOB_STATUSES.ACTIVE }], + } + + const schedule = JobSchedule.onJobFinished(prevSchedule, null) + + expect(schedule.jobs).toEqual(prevSchedule.jobs) + }) + }) }) diff --git a/src/views/connecting/Connecting.js b/src/views/connecting/Connecting.js index 15fa5c2075..e39b87d200 100644 --- a/src/views/connecting/Connecting.js +++ b/src/views/connecting/Connecting.js @@ -32,12 +32,7 @@ import { isConnectComboJobsEnabled } from 'src/redux/reducers/userFeaturesSlice' import { ErrorStatuses, ReadableStatuses } from 'src/const/Statuses' -import { - connectComplete, - initializeJobSchedule, - jobComplete, - ActionTypes, -} from 'src/redux/actions/Connect' +import { connectComplete, initializeJobSchedule, jobComplete } from 'src/redux/actions/Connect' import PostMessage from 'src/utilities/PostMessage' import { fadeOut } from 'src/utilities/Animation' @@ -51,6 +46,7 @@ import { Stack } from '@mui/material' import { usePollMember } from 'src/hooks/usePollMember' export const CONNECTING_TIMEOUT_MS = 60000 +export const MAX_FOREIGN_JOB_RETRIES = 5 export const Connecting = (props) => { const { @@ -89,6 +85,9 @@ export const Connecting = (props) => { const [message, setMessage] = useState(CONNECTING_MESSAGES.STARTING) const [timedOut, setTimedOut] = useState(false) const [connectingError, setConnectingError] = useState(null) + const [activeJobAttempt, setActiveJobAttempt] = useState(0) + const foreignJobRetriesRef = useRef(0) + const initialDataReadySentRef = useRef(false) const pollMember = usePollMember() @@ -132,7 +131,8 @@ export const Connecting = (props) => { onPostMessage('connect/memberStatusUpdate', eventData) } - if (pollingState.initialDataReady) { + if (pollingState.initialDataReady && !initialDataReadySentRef.current) { + initialDataReadySentRef.current = true // Deprecated: send initial data ready post message Oct 17, 2025 onPostMessage('connect/initialDataReady', { member_guid: pollingState.currentResponse?.member?.guid, @@ -185,10 +185,10 @@ export const Connecting = (props) => { const memberUseCasesWereProvidedInConfig = () => Boolean(connectConfig?.use_cases?.length) /** - * @returns true if currentUseCases doesn't have all the newUseCases + * @returns true if the member's use cases don't include all the configured ones */ - const memberIsMissingAConfiguredUseCase = () => { - const currentUseCases = currentMember?.use_cases + const memberIsMissingAConfiguredUseCase = (member) => { + const currentUseCases = member?.use_cases if (!currentUseCases || !Array.isArray(currentUseCases)) { return true @@ -199,55 +199,54 @@ export const Connecting = (props) => { return newUseCases.some((useCase) => currentUseCases.includes(useCase) === false) } - // When we mount, try to initialize the jobSchedule, but first we need the - // most recent job details + const loadMostRecentJob = (member) => { + if (!member?.most_recent_job_guid) return of(null) + + return defer(() => api.loadJob(member.most_recent_job_guid)).pipe( + // I have to retry here because sometimes this is too fast in sand and + // it 404s. This is a long standing backend problem. + retry(1), + // If we do error for real, just act as if there is no job + catchError(() => of(null)), + ) + } + useEffect(() => { if (!needsToInitializeJobSchedule) return () => {} - let sub$ = null - const loadJob$ = defer(() => { - // If we have a most recent job guid, get it, otherwise, just pass null - if (currentMember.most_recent_job_guid) { - return defer(() => api.loadJob(currentMember.most_recent_job_guid)).pipe( - // I have to retry here because sometimes this is too fast in sand and - // it 404s. This is a long standing backend problem. - retry(1), - // If we do error for real, just act as if there is no job - catchError(() => of(null)), - ) - } else { - return of(null) - } - }) + const refreshMember$ = defer(() => + api.loadMemberByGuid + ? api.loadMemberByGuid(currentMember.guid, clientLocale) + : Promise.resolve(currentMember), + ).pipe(catchError(() => of(currentMember))) - if ( - memberUseCasesWereProvidedInConfig() && - (memberIsMissingAConfiguredUseCase() || - currentMember.connection_status === ReadableStatuses.PENDING) - ) { - api.updateMember({ ...currentMember }, connectConfig).then((updatedMember) => { - sub$ = loadJob$.subscribe((job) => { - if (onUpsertMember) { - onUpsertMember(updatedMember) - } + const syncUseCases = (member) => { + const needsUseCaseUpdate = + memberUseCasesWereProvidedInConfig() && + (memberIsMissingAConfiguredUseCase(member) || + member.connection_status === ReadableStatuses.PENDING) - dispatch({ - type: ActionTypes.UPDATE_MEMBER_SUCCESS, - payload: { item: updatedMember }, - }) + if (!needsUseCaseUpdate) return of(member) - return dispatch( - initializeJobSchedule(currentMember, job, connectConfig, isComboJobsEnabled), - ) - }) - }) - } else { - sub$ = loadJob$.subscribe((job) => - dispatch(initializeJobSchedule(currentMember, job, connectConfig, isComboJobsEnabled)), + return defer(() => api.updateMember({ ...member }, connectConfig)).pipe( + catchError(() => of(member)), ) } - return () => sub$?.unsubscribe() + const sub$ = refreshMember$ + .pipe( + mergeMap(syncUseCases), + mergeMap((member) => loadMostRecentJob(member).pipe(map((job) => ({ member, job })))), + ) + .subscribe(({ member, job }) => { + if (member !== currentMember && onUpsertMember) { + onUpsertMember(member) + } + + dispatch(initializeJobSchedule(member, job, connectConfig, isComboJobsEnabled)) + }) + + return () => sub$.unsubscribe() }, [needsToInitializeJobSchedule]) /** @@ -269,8 +268,10 @@ export const Connecting = (props) => { mergeMap(() => api.loadMemberByGuid(currentMember.guid, clientLocale)), catchError((error) => { - // We control the scenarios of a 409 error (job already running, or member already exists). - // We can safely continue forward if that is the error we got back. + // A 409 means a job is already running for this member (for OAuth + // members that is usually the job firefly created on the redirect). + // That is fine: we poll the member by guid below and look the finished + // job up on the polled member, so the stale copy we hold is harmless. const isSafeConflictError = error?.response?.status === 409 if (isSafeConflictError) { return of(currentMember) @@ -294,12 +295,29 @@ export const Connecting = (props) => { filter((pollingState) => pollingState.pollingIsDone), pluck('currentResponse'), take(1), - mergeMap((polledResponse) => { - const loadLatestJob$ = defer(() => api.loadJob(member.most_recent_job_guid)).pipe( - map((job) => ({ member: polledResponse.member, job })), + mergeMap((polledResponse) => + loadMostRecentJob(polledResponse.member).pipe( + map((job) => ({ + member: polledResponse.member, + job: job ?? { job_type: activeJob.type }, + })), + ), + ), + mergeMap(({ member, job }) => { + const isForeignJob = job.job_type !== activeJob.type + const isStillRunning = + member.connection_status === ReadableStatuses.CONNECTED && + member.is_being_aggregated === true + + if (!isForeignJob || !isStillRunning) return of({ member, job }) + + return pollMember(member.guid).pipe( + tap((pollingState) => handleMemberPoll(pollingState)), + map((pollingState) => pollingState.currentResponse?.member), + filter((polledMember) => polledMember?.is_being_aggregated === false), + take(1), + map((idleMember) => ({ member: idleMember, job })), ) - - return loadLatestJob$ }), ), ), @@ -312,11 +330,28 @@ export const Connecting = (props) => { // if we are in an error state, fade out to ease the transition away // from this view if (ErrorStatuses.includes(member.connection_status)) { - return fadeOut(connectingRef.current, 'down').then(() => { + fadeOut(connectingRef.current, 'down').then(() => { dispatch(jobComplete(member, job, connectConfig.mode)) }) - } else { - return dispatch(jobComplete(member, job, connectConfig.mode)) + return + } + + const isForeignJob = job.job_type !== activeJob.type + const memberIsConnected = member.connection_status === ReadableStatuses.CONNECTED + + if (!isForeignJob) { + foreignJobRetriesRef.current = 0 + } else if (memberIsConnected && foreignJobRetriesRef.current >= MAX_FOREIGN_JOB_RETRIES) { + foreignJobRetriesRef.current = 0 + dispatch(jobComplete(member, { job_type: activeJob.type }, connectConfig.mode)) + return + } + + dispatch(jobComplete(member, job, connectConfig.mode)) + + if (isForeignJob && memberIsConnected) { + foreignJobRetriesRef.current += 1 + setActiveJobAttempt((attempt) => attempt + 1) } }) @@ -324,7 +359,7 @@ export const Connecting = (props) => { pollingStartedAtRef.current = null connectMember$.unsubscribe() } - }, [needsToInitializeJobSchedule, activeJob]) + }, [needsToInitializeJobSchedule, activeJob, activeJobAttempt]) /** * We removed the timeout step, but customer's relied on the timeout value in diff --git a/src/views/connecting/__tests__/ConnectingOAuthJobs-test.tsx b/src/views/connecting/__tests__/ConnectingOAuthJobs-test.tsx new file mode 100644 index 0000000000..5ee22cc393 --- /dev/null +++ b/src/views/connecting/__tests__/ConnectingOAuthJobs-test.tsx @@ -0,0 +1,277 @@ +import React from 'react' +import { createTestReduxStore, render, waitFor } from 'src/utilities/testingLibrary' +import { Connecting, MAX_FOREIGN_JOB_RETRIES } from '../Connecting' +import { PostMessageContext } from 'src/ConnectWidget' +import { ApiContextTypes, ApiProvider } from 'src/context/ApiContext' +import { POST_MESSAGES } from 'src/const/postMessages' +import { ReadableStatuses } from 'src/const/Statuses' +import { JOB_TYPES } from 'src/const/consts' +import { VERIFY_MODE } from 'src/const/Connect' + +/** + * CT-2495: after OAuth the widget lands on Connecting holding the member it + * created before the user left for the institution. That copy is PENDING, is not + * being aggregated and has no most_recent_job_guid, while firefly has already + * created (and may still be running) a job on the redirect. These tests drive + * the real redux store, job schedule and member polling through the situations + * that used to leave the widget on this screen forever. + */ + +const MEMBER_GUID = 'MBR-oauth' +const USER_GUID = 'USR-1' +const REDIRECT_JOB_GUID = 'JOB-redirect' + +type Member = { + guid: string + user_guid: string + connection_status: number + is_being_aggregated: boolean + most_recent_job_guid: string | null + is_oauth: boolean +} + +type Job = { guid: string; job_type: number; async_account_data_ready?: boolean } + +const createHttpError = (status: number, message = 'Request failed') => + Object.assign(new Error(message), { response: { status } }) + +const staleOAuthMember: Member = { + guid: MEMBER_GUID, + user_guid: USER_GUID, + connection_status: ReadableStatuses.PENDING, + is_being_aggregated: false, + most_recent_job_guid: null, + is_oauth: true, +} + +const connectedMemberRunning = (jobGuid: string): Member => ({ + ...staleOAuthMember, + connection_status: ReadableStatuses.CONNECTED, + is_being_aggregated: true, + most_recent_job_guid: jobGuid, +}) + +const createStore = () => + createTestReduxStore({ + connect: { + currentMemberGuid: MEMBER_GUID, + members: [staleOAuthMember], + jobSchedule: { isInitialized: false, jobs: [] }, + location: [], + selectedInstitution: {}, + }, + experimentalFeatures: { + memberPollingMilliseconds: 10, + optOutOfEarlyUserRelease: false, + unavailableInstitutions: [], + useWebSockets: false, + }, + }) + +const createFakeBackend = ({ pollsUntilDone = 2, earlyDataRelease = false } = {}) => { + const backend = { + member: { ...staleOAuthMember } as Member, + jobs: {} as Record, + pollsWhileRunning: 0, + + startJob(guid: string, jobType: number) { + // With early data release the job reports its data as ready while it is + // still running, which makes member polling stop before the job finishes. + backend.jobs[guid] = { guid, job_type: jobType, async_account_data_ready: earlyDataRelease } + backend.member = connectedMemberRunning(guid) + }, + + loadMemberByGuid: vi.fn(async (): Promise => { + if (backend.member.is_being_aggregated) { + backend.pollsWhileRunning += 1 + + if (backend.pollsWhileRunning >= pollsUntilDone) { + backend.pollsWhileRunning = 0 + backend.member = { ...backend.member, is_being_aggregated: false } + } + } + + return backend.member + }), + + loadJob: vi.fn(async (guid: string): Promise => { + const job = backend.jobs[guid] + + if (!job) { + throw createHttpError(404) + } + + return job + }), + + runJob: vi.fn(async (jobType: number): Promise> => { + if (backend.member.is_being_aggregated) { + // Firefly returns a 409 when the member already has a running job. + throw createHttpError(409) + } + + backend.startJob(`JOB-${jobType}`, jobType) + + return {} + }), + } + + return backend +} + +const renderConnecting = ( + backend: ReturnType, + connectConfig: Record, +) => { + const onPostMessage = vi.fn() + const api = { + loadMemberByGuid: backend.loadMemberByGuid, + loadJob: backend.loadJob, + runJob: backend.runJob, + } as unknown as ApiContextTypes + + render( + + + + + , + { store: createStore() }, + ) + + return { onPostMessage } +} + +const expectMemberConnected = (onPostMessage: ReturnType) => + waitFor( + () => + expect(onPostMessage).toHaveBeenCalledWith(POST_MESSAGES.MEMBER_CONNECTED, { + user_guid: USER_GUID, + member_guid: MEMBER_GUID, + }), + { timeout: 5000 }, + ) + +describe(' after OAuth', () => { + afterEach(() => { + vi.restoreAllMocks() + }) + + it('refreshes the stale member and waits for the job firefly created on the redirect instead of starting its own', async () => { + const backend = createFakeBackend() + // By the time Connecting mounts, firefly has created a verification job. + backend.startJob(REDIRECT_JOB_GUID, JOB_TYPES.VERIFICATION) + + const { onPostMessage } = renderConnecting(backend, { + mode: VERIFY_MODE, + include_identity: true, + }) + + await expectMemberConnected(onPostMessage) + + // Verification was already running, so only identification is started by the widget. + expect(backend.runJob.mock.calls.map((call) => call[0])).toEqual([JOB_TYPES.IDENTIFICATION]) + }) + + it('recovers from a 409 when its job conflicts with the job firefly created on the redirect', async () => { + const backend = createFakeBackend() + + // The refresh still sees the pre-OAuth member, so the widget tries to start + // verification itself. Firefly has created that job in the meantime and + // rejects the duplicate. + backend.runJob.mockImplementationOnce(async () => { + backend.startJob(REDIRECT_JOB_GUID, JOB_TYPES.VERIFICATION) + throw createHttpError(409) + }) + + const { onPostMessage } = renderConnecting(backend, { + mode: VERIFY_MODE, + include_identity: true, + }) + + await expectMemberConnected(onPostMessage) + + expect(backend.runJob.mock.calls.map((call) => call[0])).toEqual([ + JOB_TYPES.VERIFICATION, + JOB_TYPES.IDENTIFICATION, + ]) + // The stale member has no job guid. Before the fix this was requested as + // GET /jobs/null, which 404s and killed the stream. + expect(backend.loadJob).not.toHaveBeenCalledWith(null) + expect(backend.loadJob).not.toHaveBeenCalledWith(undefined) + }) + + it('starts the scheduled job once a different job finishes', async () => { + const backend = createFakeBackend() + + backend.runJob.mockImplementationOnce(async () => { + backend.startJob(REDIRECT_JOB_GUID, JOB_TYPES.AGGREGATION) + throw createHttpError(409) + }) + + const { onPostMessage } = renderConnecting(backend, { mode: VERIFY_MODE }) + + await expectMemberConnected(onPostMessage) + + expect(backend.runJob.mock.calls.map((call) => call[0])).toEqual([ + // Shows verification twice because the first call failed with 409 and the 2nd call succeeded + JOB_TYPES.VERIFICATION, + JOB_TYPES.VERIFICATION, + ]) + expect(backend.jobs[`JOB-${JOB_TYPES.VERIFICATION}`]).toBeDefined() + }) + + it('waits for a still-running job to finish before starting the next scheduled job', async () => { + const backend = createFakeBackend({ pollsUntilDone: 4, earlyDataRelease: true }) + backend.startJob(REDIRECT_JOB_GUID, JOB_TYPES.VERIFICATION) + + const { onPostMessage } = renderConnecting(backend, { + mode: VERIFY_MODE, + include_identity: true, + }) + + await expectMemberConnected(onPostMessage) + + expect(backend.runJob.mock.calls.map((call) => call[0])).toEqual([JOB_TYPES.IDENTIFICATION]) + expect(backend.jobs[`JOB-${JOB_TYPES.IDENTIFICATION}`]).toBeDefined() + expect( + onPostMessage.mock.calls.filter((call) => call[0] === 'connect/initialDataReady'), + ).toHaveLength(1) + }) + + it('stops waiting after repeated foreign jobs and moves on', async () => { + const backend = createFakeBackend() + backend.jobs[REDIRECT_JOB_GUID] = { guid: REDIRECT_JOB_GUID, job_type: JOB_TYPES.AGGREGATION } + backend.member = { ...connectedMemberRunning(REDIRECT_JOB_GUID), is_being_aggregated: false } + backend.runJob.mockImplementation(async () => { + throw createHttpError(409) + }) + + const { onPostMessage } = renderConnecting(backend, { mode: VERIFY_MODE }) + + await expectMemberConnected(onPostMessage) + + expect(backend.runJob).toHaveBeenCalledTimes(1 + MAX_FOREIGN_JOB_RETRIES) + }) + + it('still finishes when the completed job cannot be loaded', async () => { + const backend = createFakeBackend() + backend.startJob(REDIRECT_JOB_GUID, JOB_TYPES.VERIFICATION) + + let idleJobLoads = 0 + backend.loadJob.mockImplementation(async (guid: string) => { + if (!backend.member.is_being_aggregated) { + idleJobLoads += 1 + if (idleJobLoads > 1) { + throw createHttpError(500) + } + } + + return backend.jobs[guid] + }) + + const { onPostMessage } = renderConnecting(backend, { mode: VERIFY_MODE }) + + await expectMemberConnected(onPostMessage) + }) +}) From 3da66d954f679cc4754af5852b27e6c4e928c8ed Mon Sep 17 00:00:00 2001 From: Craig Tingey Date: Mon, 14 Sep 2026 14:16:14 -0600 Subject: [PATCH 2/4] remove dead logic branch --- src/views/connecting/Connecting.js | 15 +------- .../__tests__/ConnectingOAuthJobs-test.tsx | 38 ++++++++----------- 2 files changed, 17 insertions(+), 36 deletions(-) diff --git a/src/views/connecting/Connecting.js b/src/views/connecting/Connecting.js index e39b87d200..8ea67a5f51 100644 --- a/src/views/connecting/Connecting.js +++ b/src/views/connecting/Connecting.js @@ -46,7 +46,6 @@ import { Stack } from '@mui/material' import { usePollMember } from 'src/hooks/usePollMember' export const CONNECTING_TIMEOUT_MS = 60000 -export const MAX_FOREIGN_JOB_RETRIES = 5 export const Connecting = (props) => { const { @@ -86,7 +85,6 @@ export const Connecting = (props) => { const [timedOut, setTimedOut] = useState(false) const [connectingError, setConnectingError] = useState(null) const [activeJobAttempt, setActiveJobAttempt] = useState(0) - const foreignJobRetriesRef = useRef(0) const initialDataReadySentRef = useRef(false) const pollMember = usePollMember() @@ -299,7 +297,7 @@ export const Connecting = (props) => { loadMostRecentJob(polledResponse.member).pipe( map((job) => ({ member: polledResponse.member, - job: job ?? { job_type: activeJob.type }, + job: job ?? polledResponse.job ?? null, })), ), ), @@ -336,21 +334,12 @@ export const Connecting = (props) => { return } - const isForeignJob = job.job_type !== activeJob.type + const isForeignJob = job ? job.job_type !== activeJob.type : true const memberIsConnected = member.connection_status === ReadableStatuses.CONNECTED - if (!isForeignJob) { - foreignJobRetriesRef.current = 0 - } else if (memberIsConnected && foreignJobRetriesRef.current >= MAX_FOREIGN_JOB_RETRIES) { - foreignJobRetriesRef.current = 0 - dispatch(jobComplete(member, { job_type: activeJob.type }, connectConfig.mode)) - return - } - dispatch(jobComplete(member, job, connectConfig.mode)) if (isForeignJob && memberIsConnected) { - foreignJobRetriesRef.current += 1 setActiveJobAttempt((attempt) => attempt + 1) } }) diff --git a/src/views/connecting/__tests__/ConnectingOAuthJobs-test.tsx b/src/views/connecting/__tests__/ConnectingOAuthJobs-test.tsx index 5ee22cc393..e7d38b9a05 100644 --- a/src/views/connecting/__tests__/ConnectingOAuthJobs-test.tsx +++ b/src/views/connecting/__tests__/ConnectingOAuthJobs-test.tsx @@ -1,6 +1,6 @@ import React from 'react' import { createTestReduxStore, render, waitFor } from 'src/utilities/testingLibrary' -import { Connecting, MAX_FOREIGN_JOB_RETRIES } from '../Connecting' +import { Connecting } from 'src/views/connecting/Connecting' import { PostMessageContext } from 'src/ConnectWidget' import { ApiContextTypes, ApiProvider } from 'src/context/ApiContext' import { POST_MESSAGES } from 'src/const/postMessages' @@ -32,8 +32,15 @@ type Member = { type Job = { guid: string; job_type: number; async_account_data_ready?: boolean } -const createHttpError = (status: number, message = 'Request failed') => - Object.assign(new Error(message), { response: { status } }) +class HttpError extends Error { + response: { status: number } + + constructor(status: number, message = 'Request failed') { + super(message) + this.name = 'HttpError' + this.response = { status } + } +} const staleOAuthMember: Member = { guid: MEMBER_GUID, @@ -98,7 +105,7 @@ const createFakeBackend = ({ pollsUntilDone = 2, earlyDataRelease = false } = {} const job = backend.jobs[guid] if (!job) { - throw createHttpError(404) + throw new HttpError(404) } return job @@ -107,7 +114,7 @@ const createFakeBackend = ({ pollsUntilDone = 2, earlyDataRelease = false } = {} runJob: vi.fn(async (jobType: number): Promise> => { if (backend.member.is_being_aggregated) { // Firefly returns a 409 when the member already has a running job. - throw createHttpError(409) + throw new HttpError(409) } backend.startJob(`JOB-${jobType}`, jobType) @@ -181,7 +188,7 @@ describe(' after OAuth', () => { // rejects the duplicate. backend.runJob.mockImplementationOnce(async () => { backend.startJob(REDIRECT_JOB_GUID, JOB_TYPES.VERIFICATION) - throw createHttpError(409) + throw new HttpError(409) }) const { onPostMessage } = renderConnecting(backend, { @@ -206,7 +213,7 @@ describe(' after OAuth', () => { backend.runJob.mockImplementationOnce(async () => { backend.startJob(REDIRECT_JOB_GUID, JOB_TYPES.AGGREGATION) - throw createHttpError(409) + throw new HttpError(409) }) const { onPostMessage } = renderConnecting(backend, { mode: VERIFY_MODE }) @@ -239,21 +246,6 @@ describe(' after OAuth', () => { ).toHaveLength(1) }) - it('stops waiting after repeated foreign jobs and moves on', async () => { - const backend = createFakeBackend() - backend.jobs[REDIRECT_JOB_GUID] = { guid: REDIRECT_JOB_GUID, job_type: JOB_TYPES.AGGREGATION } - backend.member = { ...connectedMemberRunning(REDIRECT_JOB_GUID), is_being_aggregated: false } - backend.runJob.mockImplementation(async () => { - throw createHttpError(409) - }) - - const { onPostMessage } = renderConnecting(backend, { mode: VERIFY_MODE }) - - await expectMemberConnected(onPostMessage) - - expect(backend.runJob).toHaveBeenCalledTimes(1 + MAX_FOREIGN_JOB_RETRIES) - }) - it('still finishes when the completed job cannot be loaded', async () => { const backend = createFakeBackend() backend.startJob(REDIRECT_JOB_GUID, JOB_TYPES.VERIFICATION) @@ -263,7 +255,7 @@ describe(' after OAuth', () => { if (!backend.member.is_being_aggregated) { idleJobLoads += 1 if (idleJobLoads > 1) { - throw createHttpError(500) + throw new HttpError(500) } } From bea814f1caf7ba22ee2a7c710e93a96587cac236 Mon Sep 17 00:00:00 2001 From: Craig Tingey Date: Mon, 14 Sep 2026 16:58:40 -0600 Subject: [PATCH 3/4] fix flaky test and add null guard on job check. --- src/views/connecting/Connecting.js | 2 +- src/views/credentials/CreateMemberForm-test.tsx | 7 ++----- src/views/credentials/UpdateMemberForm-test.tsx | 7 ++----- 3 files changed, 5 insertions(+), 11 deletions(-) diff --git a/src/views/connecting/Connecting.js b/src/views/connecting/Connecting.js index 8ea67a5f51..723ab73448 100644 --- a/src/views/connecting/Connecting.js +++ b/src/views/connecting/Connecting.js @@ -302,7 +302,7 @@ export const Connecting = (props) => { ), ), mergeMap(({ member, job }) => { - const isForeignJob = job.job_type !== activeJob.type + const isForeignJob = job ? job.job_type !== activeJob.type : true const isStillRunning = member.connection_status === ReadableStatuses.CONNECTED && member.is_being_aggregated === true diff --git a/src/views/credentials/CreateMemberForm-test.tsx b/src/views/credentials/CreateMemberForm-test.tsx index 074980464b..6b63b2ea08 100644 --- a/src/views/credentials/CreateMemberForm-test.tsx +++ b/src/views/credentials/CreateMemberForm-test.tsx @@ -109,11 +109,8 @@ describe('', () => { }, }) - await waitFor(() => { - expect(screen.queryByText('Continue')).not.toBeInTheDocument() - }) - - expect(screen.getByTestId('institution-block')).toBeInTheDocument() + expect(await screen.findByTestId('institution-block')).toBeInTheDocument() + expect(screen.queryByText('Continue')).not.toBeInTheDocument() }) }) diff --git a/src/views/credentials/UpdateMemberForm-test.tsx b/src/views/credentials/UpdateMemberForm-test.tsx index fd399e328a..d897702c4b 100644 --- a/src/views/credentials/UpdateMemberForm-test.tsx +++ b/src/views/credentials/UpdateMemberForm-test.tsx @@ -106,11 +106,8 @@ describe('', () => { }, }) - await waitFor(() => { - expect(screen.queryByText('Continue')).not.toBeInTheDocument() - }) - - expect(screen.getByTestId('institution-block')).toBeInTheDocument() + expect(await screen.findByTestId('institution-block')).toBeInTheDocument() + expect(screen.queryByText('Continue')).not.toBeInTheDocument() }) }) From f52631aac2162478482510efc1446c607a039a66 Mon Sep 17 00:00:00 2001 From: Logan Rasmussen Date: Tue, 15 Sep 2026 10:39:29 -0600 Subject: [PATCH 4/4] CT-2495 connecting stall edits for conversation (#387) * docs: plan for replacing the Connecting job patches with a bounded reconcile loop * refactor: run the Connecting job schedule as a bounded reconcile loop Replace the isForeignJob / nested-poll / activeJobAttempt trio in Connecting with runJobSchedule$, a pure observable that treats the backend as the source of truth: observe whatever is running, reconcile it against the schedule, start the next job, repeat. 409s and Firefly's redirect job go through the same path. The loop is capped at jobs.length + 3 iterations and errors with JobScheduleExhaustedError, which Connecting throws to the host error boundary. Previously a perpetual 409 (or an unloadable job) re-ran runJob every poll interval forever. Integration tests added for: a foreign job followed by every scheduled job, early data release still applying when nothing else is scheduled, and bounded termination under a backend that always rejects the job. * docs: CT-2495 reconcile loop summary, risks and next steps * chore: drop planning docs from the repo * refactor: trim comments that restate the code in the job schedule loop --- src/utilities/runJobSchedule.js | 181 ++++++++++++++++++ src/views/connecting/Connecting.js | 113 +++-------- .../__tests__/ConnectingOAuthJobs-test.tsx | 109 ++++++++++- 3 files changed, 306 insertions(+), 97 deletions(-) create mode 100644 src/utilities/runJobSchedule.js diff --git a/src/utilities/runJobSchedule.js b/src/utilities/runJobSchedule.js new file mode 100644 index 0000000000..9b83d5a385 --- /dev/null +++ b/src/utilities/runJobSchedule.js @@ -0,0 +1,181 @@ +import { concat, defer, EMPTY, of, throwError } from 'rxjs' +import { catchError, filter, map, mergeMap, retry, take, tap } from 'rxjs/operators' + +import * as JobSchedule from 'src/utilities/JobSchedule' +import { JOB_STATUSES } from 'src/const/consts' +import { ReadableStatuses } from 'src/const/Statuses' + +/** + * How many extra loop iterations we allow beyond one per scheduled job. Each + * extra iteration is a job we did not start (Firefly's redirect job, a 409 race) + * that we observed and reconciled against the schedule. + */ +export const EXTRA_ITERATIONS_ALLOWED = 3 + +export class JobScheduleExhaustedError extends Error { + constructor(iterations, schedule) { + const remaining = schedule.jobs + .filter((job) => job.status !== JOB_STATUSES.DONE) + .map((job) => job.type) + .join(', ') + + super(`Gave up running the job schedule after ${iterations} attempts; remaining: ${remaining}`) + this.name = 'JobScheduleExhaustedError' + this.schedule = schedule + } +} + +const isSafeConflictError = (error) => error?.response?.status === 409 + +const isConnectedWithoutError = (member) => + member?.connection_status === ReadableStatuses.CONNECTED && !member?.error?.error_code + +/** + * Work out which job just finished, in order of trust: + * - the job we loaded fresh off the polled member + * - the job the poller itself loaded (may lack job_type over websockets) + * - if we started the job ourselves and could not load it, assume it was ours + */ +const resolveFinishedJob = (loadedJob, polledJob, startedType) => { + if (loadedJob?.job_type !== undefined) return loadedJob + if (polledJob?.job_type !== undefined) return polledJob + if (startedType !== null && startedType !== undefined) { + return { ...(loadedJob || polledJob || {}), job_type: startedType } + } + + return loadedJob ?? polledJob ?? null +} + +/** + * Drive a job schedule to completion against the backend, treating the backend + * as the source of truth for what is actually running. + * + * The widget is not the only thing that starts jobs on a member: Firefly starts + * one on the OAuth redirect, and a race can produce a 409 from `runJob`. Rather + * than special-casing "foreign" jobs, every loop iteration does the same thing: + * + * 1. If the member is being aggregated (by anyone), poll until that job is + * observable, then mark its type DONE in the schedule if it matches. + * 2. Otherwise start the schedule's active job. A 409 just means someone beat + * us to it; go back to 1 and observe. + * 3. Stop when every job is DONE, when the member leaves the CONNECTED state + * (MFA / error – the caller routes on it), or when we hit the iteration cap. + * + * Emits `{ member, job }` once per observed job completion so the caller can + * dispatch `jobComplete` and keep the progress UI in sync. Errors with + * `JobScheduleExhaustedError` if the cap is hit, or with whatever `runJob` + * rejected with for non-409 failures. + * + * @param {Object} deps + * @param {Object} deps.api needs runJob and loadJob + * @param {Function} deps.pollMember from usePollMember() + * @param {Object} deps.member the member to run jobs against (fresh) + * @param {Object} deps.schedule an initialized JobSchedule + * @param {Object} deps.config connect config passed through to runJob + * @param {Function} [deps.onPoll] called with every polling state (UI messaging, timeout) + * @return {Observable<{ member: Object, job: Object|null }>} + */ +export const runJobSchedule$ = ({ + api, + pollMember, + member, + schedule, + config, + onPoll = () => {}, +}) => { + const maxIterations = schedule.jobs.length + EXTRA_ITERATIONS_ALLOWED + + const loadJob = (memberToLoad) => { + if (!memberToLoad?.most_recent_job_guid) return of(null) + + return defer(() => api.loadJob(memberToLoad.most_recent_job_guid)).pipe( + // Sometimes this is too fast in sand and it 404s. Long standing backend problem. + retry(1), + catchError(() => of(null)), + ) + } + + /** + * Poll until the member polling logic says the UI may move on, then decide + * whether the *schedule* may move on. Early data release stops polling while + * the job is still running; that is only acceptable when there is nothing + * else scheduled after it, otherwise we keep polling until the member is idle + * so the next job can be started. + */ + const observeRunningJob = (memberGuid, currentSchedule, startedType) => + pollMember(memberGuid).pipe( + tap(onPoll), + filter((pollingState) => pollingState.pollingIsDone), + take(1), + map((pollingState) => pollingState.currentResponse), + mergeMap((polledResponse) => + loadJob(polledResponse.member).pipe( + map((job) => ({ + member: polledResponse.member, + job: resolveFinishedJob(job, polledResponse.job, startedType), + })), + ), + ), + mergeMap(({ member: polledMember, job }) => { + const hasMoreWork = !JobSchedule.areAllJobsDone( + JobSchedule.onJobFinished(currentSchedule, job), + ) + const stillRunning = + isConnectedWithoutError(polledMember) && polledMember.is_being_aggregated === true + + if (!hasMoreWork || !stillRunning) return of({ member: polledMember, job }) + + return pollMember(memberGuid).pipe( + tap(onPoll), + map((pollingState) => pollingState.currentResponse?.member), + filter((m) => m?.is_being_aggregated === false), + take(1), + map((idleMember) => ({ member: idleMember, job })), + ) + }), + ) + + const observeThenContinue = (memberGuid, currentSchedule, iteration, startedType) => + observeRunningJob(memberGuid, currentSchedule, startedType).pipe( + mergeMap(({ member: observedMember, job }) => { + const emitted = of({ member: observedMember, job }) + + if (!isConnectedWithoutError(observedMember)) return emitted + + const nextSchedule = JobSchedule.onJobFinished(currentSchedule, job) + + return concat(emitted, iterate(observedMember, nextSchedule, iteration + 1)) + }), + ) + + const iterate = (currentMember, currentSchedule, iteration) => + defer(() => { + if (JobSchedule.areAllJobsDone(currentSchedule)) return EMPTY + + if (iteration > maxIterations) { + return throwError(() => new JobScheduleExhaustedError(iteration - 1, currentSchedule)) + } + + if (currentMember.is_being_aggregated !== false) { + return observeThenContinue(currentMember.guid, currentSchedule, iteration, null) + } + + const activeJob = JobSchedule.getActiveJob(currentSchedule) + + return defer(() => api.runJob(activeJob.type, currentMember.guid, config, true)).pipe( + map(() => activeJob.type), + catchError((error) => { + // 409 is usually the job Firefly created on the OAuth redirect. + // It gets observed and reconciled like any other running job. + if (isSafeConflictError(error)) return of(null) + + return throwError(() => error) + }), + mergeMap((startedType) => + observeThenContinue(currentMember.guid, currentSchedule, iteration, startedType), + ), + ) + }) + + return iterate(member, schedule, 1) +} diff --git a/src/views/connecting/Connecting.js b/src/views/connecting/Connecting.js index 723ab73448..8263c7a973 100644 --- a/src/views/connecting/Connecting.js +++ b/src/views/connecting/Connecting.js @@ -1,17 +1,7 @@ import React, { useEffect, useState, useRef, useContext, useMemo } from 'react' import PropTypes from 'prop-types' import { defer, of } from 'rxjs' -import { - filter, - take, - pluck, - tap, - mergeMap, - concatMap, - catchError, - map, - retry, -} from 'rxjs/operators' +import { mergeMap, catchError, map, retry } from 'rxjs/operators' import { useSelector, useDispatch } from 'react-redux' import { Text } from '@mxenabled/mxui' @@ -23,6 +13,7 @@ import { CONNECTING_MESSAGES } from 'src/utilities/pollers' import { STEPS } from 'src/const/Connect' import { ProgressBar } from 'src/views/connecting/progress/ProgressBar' import * as JobSchedule from 'src/utilities/JobSchedule' +import { runJobSchedule$ } from 'src/utilities/runJobSchedule' import { AriaLive } from 'src/components/AriaLive' import { PoweredByFooter } from 'src/components/PoweredByFooter' import useAnalyticsPath from 'src/hooks/useAnalyticsPath' @@ -84,12 +75,10 @@ export const Connecting = (props) => { const [message, setMessage] = useState(CONNECTING_MESSAGES.STARTING) const [timedOut, setTimedOut] = useState(false) const [connectingError, setConnectingError] = useState(null) - const [activeJobAttempt, setActiveJobAttempt] = useState(0) const initialDataReadySentRef = useRef(false) const pollMember = usePollMember() - const activeJob = JobSchedule.getActiveJob(jobSchedule) const needsToInitializeJobSchedule = jobSchedule.isInitialized === false function handleMemberPoll(pollingState) { @@ -248,79 +237,25 @@ export const Connecting = (props) => { }, [needsToInitializeJobSchedule]) /** - * If the member is not aggregating, start a job, otherwise, poll the - * member until it's done aggregating. + * Runs once per schedule initialization rather than once per active job: + * runJobSchedule$ tracks the schedule itself (including jobs Firefly started + * and 409 races) and redux is kept in step through jobComplete, which applies + * the same JobSchedule.onJobFinished. */ useEffect(() => { - // If we still need to initialize the job schedule, do nothing - if (needsToInitializeJobSchedule || !activeJob) return () => {} + if (needsToInitializeJobSchedule || !JobSchedule.getActiveJob(jobSchedule)) return () => {} pollingStartedAtRef.current = Date.now() - const connectMember$ = defer(() => { - const needsJobStarted = currentMember.is_being_aggregated === false - - const startJob$ = defer(() => - api.runJob(activeJob?.type, currentMember.guid, connectConfig, true), - ).pipe( - mergeMap(() => api.loadMemberByGuid(currentMember.guid, clientLocale)), - - catchError((error) => { - // A 409 means a job is already running for this member (for OAuth - // members that is usually the job firefly created on the redirect). - // That is fine: we poll the member by guid below and look the finished - // job up on the polled member, so the stale copy we hold is harmless. - const isSafeConflictError = error?.response?.status === 409 - if (isSafeConflictError) { - return of(currentMember) - } - - // Prevent the Connecting component from trying to continue - // when a bad error occurs. - setConnectingError(error) - throw error - }), - ) - - // If the current member is not being aggregated, start a job - // otherwise, just go with the member we have now - return needsJobStarted ? startJob$ : of(currentMember) - }) - .pipe( - concatMap((member) => - pollMember(member.guid).pipe( - tap((pollingState) => handleMemberPoll(pollingState)), - filter((pollingState) => pollingState.pollingIsDone), - pluck('currentResponse'), - take(1), - mergeMap((polledResponse) => - loadMostRecentJob(polledResponse.member).pipe( - map((job) => ({ - member: polledResponse.member, - job: job ?? polledResponse.job ?? null, - })), - ), - ), - mergeMap(({ member, job }) => { - const isForeignJob = job ? job.job_type !== activeJob.type : true - const isStillRunning = - member.connection_status === ReadableStatuses.CONNECTED && - member.is_being_aggregated === true - - if (!isForeignJob || !isStillRunning) return of({ member, job }) - - return pollMember(member.guid).pipe( - tap((pollingState) => handleMemberPoll(pollingState)), - map((pollingState) => pollingState.currentResponse?.member), - filter((polledMember) => polledMember?.is_being_aggregated === false), - take(1), - map((idleMember) => ({ member: idleMember, job })), - ) - }), - ), - ), - ) - .subscribe(({ member, job }) => { + const schedule$ = runJobSchedule$({ + api, + pollMember, + member: currentMember, + schedule: jobSchedule, + config: connectConfig, + onPoll: handleMemberPoll, + }).subscribe({ + next: ({ member, job }) => { if (onUpsertMember) { onUpsertMember(member) } @@ -334,21 +269,17 @@ export const Connecting = (props) => { return } - const isForeignJob = job ? job.job_type !== activeJob.type : true - const memberIsConnected = member.connection_status === ReadableStatuses.CONNECTED - dispatch(jobComplete(member, job, connectConfig.mode)) - - if (isForeignJob && memberIsConnected) { - setActiveJobAttempt((attempt) => attempt + 1) - } - }) + }, + // Thrown from render below so the host's error boundary takes over. + error: (error) => setConnectingError(error), + }) return () => { pollingStartedAtRef.current = null - connectMember$.unsubscribe() + schedule$.unsubscribe() } - }, [needsToInitializeJobSchedule, activeJob, activeJobAttempt]) + }, [needsToInitializeJobSchedule]) /** * We removed the timeout step, but customer's relied on the timeout value in diff --git a/src/views/connecting/__tests__/ConnectingOAuthJobs-test.tsx b/src/views/connecting/__tests__/ConnectingOAuthJobs-test.tsx index e7d38b9a05..7589204f9b 100644 --- a/src/views/connecting/__tests__/ConnectingOAuthJobs-test.tsx +++ b/src/views/connecting/__tests__/ConnectingOAuthJobs-test.tsx @@ -7,6 +7,7 @@ import { POST_MESSAGES } from 'src/const/postMessages' import { ReadableStatuses } from 'src/const/Statuses' import { JOB_TYPES } from 'src/const/consts' import { VERIFY_MODE } from 'src/const/Connect' +import { EXTRA_ITERATIONS_ALLOWED } from 'src/utilities/runJobSchedule' /** * CT-2495: after OAuth the widget lands on Connecting holding the member it @@ -126,11 +127,35 @@ const createFakeBackend = ({ pollsUntilDone = 2, earlyDataRelease = false } = {} return backend } +/** + * Connecting throws `connectingError` during render so the host's error + * boundary can take over. Tests need a boundary of their own to observe that. + */ +class TestErrorBoundary extends React.Component< + { onError: (error: Error) => void; children: React.ReactNode }, + { hasError: boolean } +> { + state = { hasError: false } + + componentDidCatch(error: Error) { + this.props.onError(error) + } + + static getDerivedStateFromError() { + return { hasError: true } + } + + render() { + return this.state.hasError ?
: this.props.children + } +} + const renderConnecting = ( backend: ReturnType, connectConfig: Record, ) => { const onPostMessage = vi.fn() + const onError = vi.fn() const api = { loadMemberByGuid: backend.loadMemberByGuid, loadJob: backend.loadJob, @@ -138,15 +163,17 @@ const renderConnecting = ( } as unknown as ApiContextTypes render( - - - - - , + + + + + + + , { store: createStore() }, ) - return { onPostMessage } + return { onPostMessage, onError } } const expectMemberConnected = (onPostMessage: ReturnType) => @@ -246,6 +273,76 @@ describe(' after OAuth', () => { ).toHaveLength(1) }) + it('runs every scheduled job after a foreign job that was already running', async () => { + const backend = createFakeBackend() + // Firefly kicked off a plain aggregation on the redirect; the widget wants + // verification + identity. All three must run, in order, exactly once. + backend.startJob(REDIRECT_JOB_GUID, JOB_TYPES.AGGREGATION) + + const { onPostMessage } = renderConnecting(backend, { + mode: VERIFY_MODE, + include_identity: true, + }) + + await expectMemberConnected(onPostMessage) + + expect(backend.runJob.mock.calls.map((call) => call[0])).toEqual([ + JOB_TYPES.VERIFICATION, + JOB_TYPES.IDENTIFICATION, + ]) + }) + + it('still releases the user early when the running job satisfies the schedule', async () => { + // Early data release is a product feature: once the job reports its data is + // ready we hand off before aggregation finishes. Waiting to idle must only + // happen when there is more scheduled work to do. + const backend = createFakeBackend({ pollsUntilDone: 1000, earlyDataRelease: true }) + backend.startJob(REDIRECT_JOB_GUID, JOB_TYPES.VERIFICATION) + + const { onPostMessage } = renderConnecting(backend, { mode: VERIFY_MODE }) + + await expectMemberConnected(onPostMessage) + + expect(backend.member.is_being_aggregated).toBe(true) + expect(backend.runJob).not.toHaveBeenCalled() + }) + + it('gives up with an error instead of retrying forever when the backend keeps rejecting the job', async () => { + const backend = createFakeBackend() + // The member reports idle but every runJob is rejected as a conflict, so no + // iteration can ever make progress. + backend.jobs[REDIRECT_JOB_GUID] = { guid: REDIRECT_JOB_GUID, job_type: JOB_TYPES.AGGREGATION } + backend.member = { + ...staleOAuthMember, + connection_status: ReadableStatuses.CONNECTED, + is_being_aggregated: false, + most_recent_job_guid: REDIRECT_JOB_GUID, + } + backend.runJob.mockImplementation(async () => { + throw new HttpError(409) + }) + + vi.spyOn(console, 'error').mockImplementation(() => {}) + + const { onPostMessage, onError } = renderConnecting(backend, { mode: VERIFY_MODE }) + + await waitFor(() => expect(onError).toHaveBeenCalled(), { timeout: 5000 }) + expect(onError.mock.calls[0][0].name).toBe('JobScheduleExhaustedError') + + // One attempt per scheduled job plus a small allowance for jobs we did not + // start. With a 10ms poll interval the old code made hundreds of calls here. + const scheduledJobs = 1 + expect(backend.runJob).toHaveBeenCalledTimes(scheduledJobs + EXTRA_ITERATIONS_ALLOWED) + expect(onPostMessage).not.toHaveBeenCalledWith( + POST_MESSAGES.MEMBER_CONNECTED, + expect.anything(), + ) + + const callsAtError = backend.runJob.mock.calls.length + await new Promise((resolve) => setTimeout(resolve, 200)) + expect(backend.runJob.mock.calls.length).toBe(callsAtError) + }) + it('still finishes when the completed job cannot be loaded', async () => { const backend = createFakeBackend() backend.startJob(REDIRECT_JOB_GUID, JOB_TYPES.VERIFICATION)