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)