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/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 15fa5c2075..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' @@ -32,12 +23,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' @@ -89,10 +75,10 @@ export const Connecting = (props) => { const [message, setMessage] = useState(CONNECTING_MESSAGES.STARTING) const [timedOut, setTimedOut] = useState(false) const [connectingError, setConnectingError] = useState(null) + const initialDataReadySentRef = useRef(false) const pollMember = usePollMember() - const activeJob = JobSchedule.getActiveJob(jobSchedule) const needsToInitializeJobSchedule = jobSchedule.isInitialized === false function handleMemberPoll(pollingState) { @@ -132,7 +118,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 +172,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,112 +186,76 @@ 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) - } - - dispatch({ - type: ActionTypes.UPDATE_MEMBER_SUCCESS, - payload: { item: updatedMember }, - }) + const syncUseCases = (member) => { + const needsUseCaseUpdate = + memberUseCasesWereProvidedInConfig() && + (memberIsMissingAConfiguredUseCase(member) || + member.connection_status === ReadableStatuses.PENDING) - return dispatch( - initializeJobSchedule(currentMember, job, connectConfig, isComboJobsEnabled), - ) - }) - }) - } else { - sub$ = loadJob$.subscribe((job) => - dispatch(initializeJobSchedule(currentMember, job, connectConfig, isComboJobsEnabled)), + if (!needsUseCaseUpdate) return of(member) + + 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]) /** - * 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) => { - // 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. - 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) => { - const loadLatestJob$ = defer(() => api.loadJob(member.most_recent_job_guid)).pipe( - map((job) => ({ member: polledResponse.member, job })), - ) - - return loadLatestJob$ - }), - ), - ), - ) - .subscribe(({ member, job }) => { + const schedule$ = runJobSchedule$({ + api, + pollMember, + member: currentMember, + schedule: jobSchedule, + config: connectConfig, + onPoll: handleMemberPoll, + }).subscribe({ + next: ({ member, job }) => { if (onUpsertMember) { onUpsertMember(member) } @@ -312,19 +263,23 @@ 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 } - }) + + dispatch(jobComplete(member, job, connectConfig.mode)) + }, + // 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]) + }, [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 new file mode 100644 index 0000000000..7589204f9b --- /dev/null +++ b/src/views/connecting/__tests__/ConnectingOAuthJobs-test.tsx @@ -0,0 +1,366 @@ +import React from 'react' +import { createTestReduxStore, render, waitFor } from 'src/utilities/testingLibrary' +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' +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 + * 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 } + +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, + 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 new HttpError(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 new HttpError(409) + } + + backend.startJob(`JOB-${jobType}`, jobType) + + return {} + }), + } + + 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, + runJob: backend.runJob, + } as unknown as ApiContextTypes + + render( + + + + + + + , + { store: createStore() }, + ) + + return { onPostMessage, onError } +} + +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 new HttpError(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 new HttpError(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('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) + + let idleJobLoads = 0 + backend.loadJob.mockImplementation(async (guid: string) => { + if (!backend.member.is_being_aggregated) { + idleJobLoads += 1 + if (idleJobLoads > 1) { + throw new HttpError(500) + } + } + + return backend.jobs[guid] + }) + + const { onPostMessage } = renderConnecting(backend, { mode: VERIFY_MODE }) + + await expectMemberConnected(onPostMessage) + }) +}) 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() }) })