Synchronous execution of disposer

Describe the bug

We implement an operator called combineMap which is a more efficient variant of the following pattern:

values$.pipe(rxjs.switchMap(values => value.length > 0 ? rxjs.combineMap(values.map(selector)) : rxjs.of([]) 

Instead, we can do the following which will re-use the results and subscriptions for unchanged values:

values$.pipe(combineMap(selector))

This works fine most of the time, but sometimes in production we get the following error:

    TypeError: Cannot read properties of null (reading 'unsubscribe')
        at /home/jesper/nxtedition/nxt/asset/node_modules/@nxtedition/lib/rxjs/combineMap.js:136:28
        at execFinalizer (/home/jesper/nxtedition/nxt/asset/node_modules/rxjs/dist/cjs/internal/Subscription.js:172:9)
        at Subscription.unsubscribe (/home/jesper/nxtedition/nxt/asset/node_modules/rxjs/dist/cjs/internal/Subscription.js:89:29)
        at Subscriber.unsubscribe (/home/jesper/nxtedition/nxt/asset/node_modules/rxjs/dist/cjs/internal/Subscriber.js:75:42)
        at OperatorSubscriber.unsubscribe (/home/jesper/nxtedition/nxt/asset/node_modules/rxjs/dist/cjs/internal/operators/OperatorSubscriber.js:72:42)
        at execFinalizer (/home/jesper/nxtedition/nxt/asset/node_modules/rxjs/dist/cjs/internal/Subscription.js:175:19)
        at Subscription.unsubscribe (/home/jesper/nxtedition/nxt/asset/node_modules/rxjs/dist/cjs/internal/Subscription.js:89:29)
        at Subscriber.unsubscribe (/home/jesper/nxtedition/nxt/asset/node_modules/rxjs/dist/cjs/internal/Subscriber.js:75:42)
        at OperatorSubscriber.unsubscribe (/home/jesper/nxtedition/nxt/asset/node_modules/rxjs/dist/cjs/internal/operators/OperatorSubscriber.js:72:42)
        at execFinalizer (/home/jesper/nxtedition/nxt/asset/node_modules/rxjs/dist/cjs/internal/Subscription.js:175:19)

Give the following code:

const EMPTY = Object.freeze([])

function combineMap(project, equals = (a, b) => a === b) {
  const self = this
  return new rxjs.Observable((o) => {
    let curr = EMPTY
    let scheduled = false
    let dirty = false
    let active = 0
    let empty = 0

    const _error = (err) => o.error(err)

    function _update() {
      scheduled = false

      if (empty) {
        return
      }

      if (dirty) {
        dirty = false
        o.next(curr.map((context) => context.value))
      }

      if (!active) {
        o.complete()
      }
    }

    function update() {
      if (!scheduled) {
        scheduled = true
        queueMicrotask(_update)
      }
    }

    active += 1
    const subscription = self.subscribe({
      next(keys) {
        if (!Array.isArray(keys)) {
          keys = EMPTY
        }
        // TODO (perf): Avoid array allocation & copy if nothing has updated.

        const prev = curr
        curr = new Array(keys.length)

        const prevLen = prev.length
        const currLen = curr.length

        if (currLen !== prevLen || prev === EMPTY) {
          dirty = true
          update()
        }

        for (let n = 0; n < currLen; ++n) {
          const key = keys[n]

          if (n < prevLen && prev[n] && equals(prev[n].key, key)) {
            curr[n] = prev[n]
            prev[n] = null
            continue
          }

          dirty = true
          update()

          // TODO (perf): Guess start index based on n, e.g. n - 1 and n + 1 to check if
          // a key has simply been added or removed.
          const i = prev.findIndex((entry) => entry && equals(entry.key, key))

          if (i !== -1) {
            curr[n] = prev[i]
            prev[i] = null
          } else {
            const entry = (curr[n] = {
              key,
              value: EMPTY,
              subscription: null,
            })

            let observable
            try {
              observable = rxjs.from(project(keys[n]))
            } catch (err) {
              observable = rxjs.throwError(() => err)
            }

            empty += 1
            active += 1

            const subscription = observable.subscribe({
              next(value) {
                if (entry.value === EMPTY) {
                  empty -= 1
                }

                entry.value = value

                dirty = true
                update()
              },
              error: _error,
            })
            // ** outer disposer runs before this assignment is done
            entry.subscription = subscription
            entry.subscription.add(() => {
              if (entry.value === EMPTY) {
                empty -= 1
              }

              active -= 1

              dirty = true
              update()
            })
          }
        }

        // TODO (perf): start from index where prev[n] is not null.
        for (let n = 0; n < prevLen; n++) {
          // ** this will crash since subscription is null
          prev[n]?.subscription.unsubscribe()
        }
      },
      error: _error,
      complete() {
        active -= 1
        if (!active) {
          update()
        }
      },
    })

    return () => {
      for (const entry of curr) {
        entry?.subscription.unsubscribe()
      }
      subscription.unsubscribe()
    }
  })
}

For some reason the disposer:

    return () => {
      for (const entry of curr) {
        entry?.subscription.unsubscribe()
      }
      subscription.unsubscribe()
    }

Runs before subscription has been assigned above:

entry.subscription = subscription

I have no idea how this can occur and have been unable to reproduce it.

Expected behavior

The outer disposer does not run concurrently with the inner subscription's next invocation.

Reproduction code

Unable to reproduce outside of production code.

Reproduction URL

No response

Version

7.8.0

Environment

No response

Additional context

No response

Start from the combineMap implementation shown in the issue and trace RxJS Subscription.unsubscribe and execFinalizer, especially the ordering around the inner observable's subscribe callback and entry.subscription assignment. Add a regression test that exercises synchronous disposal during subscription, then verify the outer disposer cannot access an unassigned inner subscription while preserving cleanup of active entries.