CT-2332 | Fix false-success during transport when websockets are turned on

docs/APIDOCUMENTATION.md

@@ -135,6 +135,8 @@
 ##### Notes
 
 > This callback is also used during OAuth flows to synchronize member data when the backend returns a different `inbound_member_guid` than the one used to start the flow (e.g., during non-OAuth to OAuth migrations). When this happens, the widget will fetch the new member record and update its internal state to use the new GUID.
+>
+> `most_recent_job_guid` must change when a new job starts and be `null` before the first one. A `CONNECTED` member whose value is `null`, or unchanged since the widget called `runJob`, keeps the Connecting step waiting.
 
 ##### Responses
 

src/utilities/__tests__/runJobSchedule-test.js

@@ -0,0 +1,139 @@
+import { Subject } from 'rxjs'
+
+import { runJobSchedule$ } from 'src/utilities/runJobSchedule'
+import { JOB_STATUSES, JOB_TYPES } from 'src/const/consts'
+import { ReadableStatuses } from 'src/const/Statuses'
+
+// CT-2332 pre-job update rules, driven through runJobSchedule$'s injected api and pollMember.
+
+const MEMBER_GUID = 'MBR-1'
+const OUR_JOB_GUID = 'JOB-1'
+
+const verificationSchedule = () => ({
+  isInitialized: true,
+  jobs: [{ type: JOB_TYPES.VERIFICATION, status: JOB_STATUSES.ACTIVE }],
+})
+
+const member = (connectionStatus, extra = {}) => ({
+  guid: MEMBER_GUID,
+  connection_status: connectionStatus,
+  is_being_aggregated: false,
+  ...extra,
+})
+
+// Over websockets the polled job carries only the guid, never a job_type.
+const doneState = (polledMember) => ({
+  pollingIsDone: true,
+  currentResponse: {
+    member: polledMember,
+    job: { guid: polledMember.most_recent_job_guid ?? null, async_account_data_ready: false },
+  },
+})
+
+const verificationJob = (guid) => ({ guid, job_type: JOB_TYPES.VERIFICATION })
+
+const flush = () => new Promise((resolve) => setTimeout(resolve, 0))
+
+const run = ({ api, member: startingMember }) => {
+  const pollingStates$ = new Subject()
+  const emissions = []
+
+  const subscription = runJobSchedule$({
+    api,
+    pollMember: () => pollingStates$,
+    member: startingMember,
+    schedule: verificationSchedule(),
+    config: {},
+  }).subscribe({ next: (emission) => emissions.push(emission) })
+
+  return { pollingStates$, emissions, subscription }
+}
+
+describe('runJobSchedule$ pre-job updates', () => {
+  it('does not treat a CONNECTED member with most_recent_job_guid null as the finished job', async () => {
+    const api = {
+      runJob: vi.fn().mockResolvedValue({}),
+      loadJob: vi.fn(async (guid) => verificationJob(guid)),
+    }
+    const { pollingStates$, emissions, subscription } = run({
+      api,
+      member: member(ReadableStatuses.PENDING, { most_recent_job_guid: null }),
+    })
+    await flush()
+
+    // Firefly's pre-job CONNECTED update, delivered late.
+    pollingStates$.next(
+      doneState(member(ReadableStatuses.CONNECTED, { most_recent_job_guid: null })),
+    )
+    await flush()
+
+    expect(emissions).toHaveLength(0)
+    expect(api.loadJob).not.toHaveBeenCalled()
+
+    const impeded = member(ReadableStatuses.IMPEDED, { most_recent_job_guid: OUR_JOB_GUID })
+    pollingStates$.next(doneState(impeded))
+    await flush()
+
+    expect(emissions).toHaveLength(1)
+    expect(emissions[0]).toMatchObject({ member: impeded, job: verificationJob(OUR_JOB_GUID) })
+
+    subscription.unsubscribe()
+  })
+
+  it('does not treat an idle CONNECTED member still naming its previous job as the finished job', async () => {
+    const returningMember = member(ReadableStatuses.CONNECTED, {
+      most_recent_job_guid: 'JOB-old',
+    })
+    const api = {
+      runJob: vi.fn().mockResolvedValue({}),
+      loadJob: vi.fn(async (guid) => verificationJob(guid)),
+    }
+    const { pollingStates$, emissions, subscription } = run({ api, member: returningMember })
+    await flush()
+
+    // The same late update for a returning member names the old job.
+    pollingStates$.next(doneState(returningMember))
+    await flush()
+
+    expect(emissions).toHaveLength(0)
+    expect(api.loadJob).not.toHaveBeenCalled()
+
+    const impeded = member(ReadableStatuses.IMPEDED, { most_recent_job_guid: OUR_JOB_GUID })
+    pollingStates$.next(doneState(impeded))
+    await flush()
+
+    expect(emissions).toHaveLength(1)
+    expect(emissions[0]).toMatchObject({ member: impeded, job: verificationJob(OUR_JOB_GUID) })
+
+    subscription.unsubscribe()
+  })
+
+  it('observes the job firefly assigned when runJob is rejected with a 409, instead of treating it as the previous job', async () => {
+    // Firefly started this job on the OAuth redirect. It is the job that matters: the
+    // widget's own runJob is a duplicate and firefly rejects it with a 409.
+    const FIREFLY_JOB_GUID = 'JOB-firefly'
+    const memberWithFireflyJob = member(ReadableStatuses.CONNECTED, {
+      most_recent_job_guid: FIREFLY_JOB_GUID,
+    })
+    const conflict = Object.assign(new Error('conflict'), { response: { status: 409 } })
+    const api = {
+      runJob: vi.fn().mockRejectedValue(conflict),
+      loadJob: vi.fn(async (guid) => verificationJob(guid)),
+    }
+    const { pollingStates$, emissions, subscription } = run({ api, member: memberWithFireflyJob })
+    await flush()
+
+    expect(api.runJob).toHaveBeenCalledTimes(1)
+
+    // The member still names firefly's job when it finishes. Had the 409 path recorded that
+    // guid as "the previous job", this update would be ignored and Connecting would hang.
+    pollingStates$.next(doneState(memberWithFireflyJob))
+    await flush()
+
+    expect(api.loadJob).toHaveBeenCalledWith(FIREFLY_JOB_GUID)
+    expect(emissions).toHaveLength(1)
+    expect(emissions[0].job).toEqual(verificationJob(FIREFLY_JOB_GUID))
+
+    subscription.unsubscribe()
+  })
+})

src/utilities/runJobSchedule.js

@@ -30,6 +30,17 @@ const isSafeConflictError = (error) => error?.response?.status === 409
 const isConnectedWithoutError = (member) =>
   member?.connection_status === ReadableStatuses.CONNECTED && !member?.error?.error_code
 
+const NOT_STARTED_BY_US = { type: null, previousJobGuid: null }
+
+// Firefly sets an OAuth member CONNECTED on the redirect before any job exists, and over
+// websockets that update can arrive after we started ours (CT-2332). It names the job the
+// member had before runJob: none for a first job, the previous job for a returning member.
+const isPreJobUpdate = (member, started) => {
+  const guid = member?.most_recent_job_guid
+
+  return !guid || guid === started.previousJobGuid
+}
+
 /**
  * Work out which job just finished, in order of trust:
  *  - the job we loaded fresh off the polled member
@@ -102,17 +113,26 @@ export const runJobSchedule$ = ({
    * 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) =>
+  const observeRunningJob = (memberGuid, currentSchedule, started) =>
     pollMember(memberGuid).pipe(
+      // onPoll runs before the gate on purpose: it is where the Connecting timeout lives.
       tap(onPoll),
-      filter((pollingState) => pollingState.pollingIsDone),
+      // Error and MFA states route on the member alone; only CONNECTED needs a real finished job.
+      filter((pollingState) => {
+        const polledMember = pollingState.currentResponse?.member
+
+        return (
+          pollingState.pollingIsDone &&
+          !(isConnectedWithoutError(polledMember) && isPreJobUpdate(polledMember, started))
+        )
+      }),
       take(1),
       map((pollingState) => pollingState.currentResponse),
       mergeMap((polledResponse) =>
         loadJob(polledResponse.member).pipe(
           map((job) => ({
             member: polledResponse.member,
-            job: resolveFinishedJob(job, polledResponse.job, startedType),
+            job: resolveFinishedJob(job, polledResponse.job, started.type),
           })),
         ),
       ),
@@ -135,8 +155,8 @@ export const runJobSchedule$ = ({
       }),
     )
 
-  const observeThenContinue = (memberGuid, currentSchedule, iteration, startedType) =>
-    observeRunningJob(memberGuid, currentSchedule, startedType).pipe(
+  const observeThenContinue = (memberGuid, currentSchedule, iteration, started) =>
+    observeRunningJob(memberGuid, currentSchedule, started).pipe(
       mergeMap(({ member: observedMember, job }) => {
         const emitted = of({ member: observedMember, job })
 
@@ -157,22 +177,27 @@ export const runJobSchedule$ = ({
       }
 
       if (currentMember.is_being_aggregated !== false) {
-        return observeThenContinue(currentMember.guid, currentSchedule, iteration, null)
+        return observeThenContinue(
+          currentMember.guid,
+          currentSchedule,
+          iteration,
+          NOT_STARTED_BY_US,
+        )
       }
 
       const activeJob = JobSchedule.getActiveJob(currentSchedule)
 
       return defer(() => api.runJob(activeJob.type, currentMember.guid, config, true)).pipe(
-        map(() => activeJob.type),
+        map(() => ({ type: activeJob.type, previousJobGuid: currentMember.most_recent_job_guid })),
         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)
+          if (isSafeConflictError(error)) return of(NOT_STARTED_BY_US)
 
           return throwError(() => error)
         }),
-        mergeMap((startedType) =>
-          observeThenContinue(currentMember.guid, currentSchedule, iteration, startedType),
+        mergeMap((started) =>
+          observeThenContinue(currentMember.guid, currentSchedule, iteration, started),
         ),
       )
     })