CT-2495 connecting stall edits for conversation

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

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