JobScheduler and Connecting know to expect a job started from firefly

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) => {

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', () => {

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) => {

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)
+    })
+  })
 })

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)
+}