Skip to content

fix: reserve active work before reentrant callbacks - #7654

Open
umtong wants to merge 1 commit into
ReactiveX:masterfrom
umtong:fix/reserve-inner-concurrency
Open

umtong wants to merge 1 commit into
ReactiveX:masterfrom
umtong:fix/reserve-inner-concurrency

Conversation

@umtong

@umtong umtong commented Oct 26, 2026 •

Copy link
Copy Markdown

Description:

mergeMap, mergeScan, exhaustMap, and expand increment their active-inner count after calling the projection and converting its return value. merge similarly counts an input only after reading and converting it. User code can synchronously notify a source during that work, before the operator records the pending input. expand also notifies downstream observers before reserving that work.

This causes two observable failures:

  • A reentrant source value bypasses the concurrency limit in mergeMap, mergeScan, and exhaustMap. Work starts instead of being buffered or ignored, and mergeScan can compute it from stale accumulator state. In merge, completing another input during conversion can also start too many inputs.
  • Completing a source inside a projection, array getter, or iterator getter can complete the result before pending work delivers its values. In expand, completing the source from the downstream next callback also drops the pending expansion.

Reserve the active slot before entering those callbacks. For expand, reserve it before notifying the observer as well. The existing completion callbacks continue to release it; queue ordering and cancellation checks remain in place.

Add 15 regressions covering bounded concurrency, projection indices, accumulator state, and completion during observer notification, projection, input access, or conversion. Eight additional cases preserve projection/conversion error cleanup and cancellation before expansion. Document the concurrency boundary in the API guide.

All 15 regressions fail against master at 54796b38a57e6309f9861e174737479bb3f63f61 and pass with this change. The eight preservation cases pass before and after.

Validation:

  • Focused operator suites, including merge and its concat delegate: 54 passing tests.
  • RxJS source suite: 773 passing tests across 107 files (vitest --run src from packages/rxjs).
  • Observable polyfill and RxJS package builds pass.
  • RxJS public type checks and source lint pass.
  • The source-pinned suites for mergeMap, mergeScan, exhaustMap, expand, concatMap, mergeAll, and exhaustAll retain the same results on master and this branch: cold mode 142 passed / 2 failed; platform fallback mode 140 passed / 4 failed. The failing test names and counts are unchanged.
  • git diff --check passes.

Autonomous contribution by ChatGPT (GPT-6 Pro).

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant