From e267b44d75a20124563e1ee545619b6f88e71e1e Mon Sep 17 00:00:00 2001 From: XperiAndri Date: Mon, 5 Oct 2026 02:12:28 +0200 Subject: [PATCH] fix(rxquery): emit `yield`/`zero` synchronously to keep element order `Yield` and `Zero` used `Observable.Return/Empty (TimeProvider.System)`, which R3 schedules on the thread pool. Every element of a query therefore hopped to another thread, `SelectMany` merged them in arbitrary order and the results arrived after the source had moved on, so `take`, `head`, `last`, `zip` and friends were nondeterministic. Both now emit synchronously on subscription. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 4 ++++ src/FSharp.Control.R3/Observable.fs | 6 ++++-- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4cc6deb..df56a03 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Breaking:** `AwaitOperationConfiguration` cases are prefixed with `Await` (`AwaitSequential`, `AwaitParallel 4`, ...) so they no longer collide with `System.Threading.Tasks.Parallel` - `mapAsync` and `iterAsync` validate the concurrency limit of the options when they are called +### Fixed + +- `rxquery` emitted its elements on the thread pool, out of order and after the source had moved on; `yield` and `zero` are now synchronous + ## [0.3.1] - 2026-01-28 ### Fixed diff --git a/src/FSharp.Control.R3/Observable.fs b/src/FSharp.Control.R3/Observable.fs index b728290..984f385 100644 --- a/src/FSharp.Control.R3/Observable.fs +++ b/src/FSharp.Control.R3/Observable.fs @@ -131,8 +131,10 @@ module Builders = member _.SkipWhile (s : Observable<_>, [] predicate : _ -> bool) = s.SkipWhile (predicate) [] member _.Skip (s : Observable<_>, count : int) = s.Skip (count) - member _.Zero () = Observable.Empty (TimeProvider.System) - member _.Yield (value) = Observable.Return (value, TimeProvider.System) + // Zero and Yield emit synchronously on subscription: scheduling them on TimeProvider.System hopped every element of + // a query to the thread pool, so SelectMany merged them in arbitrary order and the results arrived late + member _.Zero () : Observable<'T> = Observable.Empty<'T>() + member _.Yield (value : 'T) = Observable.Return<'T> value [] member _.Count (s : Observable<_>) = ObservableExtensions.CountAsync (s) []