Skip to content

CLAUDE.md

Latest commit

 

History

History
120 lines (93 loc) · 48.3 KB

CLAUDE.md

File metadata and controls

120 lines (93 loc) · 48.3 KB
Raw

ReactiveExtensionsSharp

A .NET port of RxJS. C# namespace: ReactiveExtensionsSharp. NuGet PackageId: ReactiveExtensionsSharp (the plain ReactiveExtensionsSharp package name was already taken on nuget.org). GitHub repo: hardkoded/ReactiveExtensions-Sharp (directory on disk may still be named rxjs-sharp).

Why this exists

Upstream Puppeteer (JS) uses rxjs internally for timeout/retry/cancellation combinators (Locator actions, network-idle detection, screencast throttling). The long-term goal is to make this port usable inside puppeteer-sharp for the same purpose. See "Puppeteer-essential surface" below for the priority list this implies.

Workflow rules

  • TDD, always: for every primitive/operator, write the failing NUnit test first, then implement until it passes.
  • One branch, one PR, per feature: each operator group / primitive / milestone slice lands as its own branch + PR merged to main. No large multi-feature branches.
  • Honor the original RxJS test suite: don't invent test cases when an upstream one already exists. Port from RxJS tag 7.8.2 (the last stable pre-rewrite release — HEAD of the rxjs repo is an unstable 8.0 alpha with a different architecture). Upstream repo: /Users/dario/Code/reactivex/rxjs. Spec files live at the repo root under spec/operators/*-spec.ts and spec/observables/*-spec.ts at that tag (not under packages/rxjs/spec/ — that nested layout only exists on the HEAD monorepo rewrite). To browse the tag without disturbing whatever branch is currently checked out there, use a throwaway worktree: 
    git -C /Users/dario/Code/reactivex/rxjs worktree add --detach /tmp/rxjs-782-ref 7.8.2
# ... read files under /tmp/rxjs-782-ref/spec/ ...
git -C /Users/dario/Code/reactivex/rxjs worktree remove /tmp/rxjs-782-ref
  • Puppeteer usage must show up in tests, not just a one-off playground check: test/ReactiveExtensionsSharp.Tests/PuppeteerScenarios/ holds hand-written cases (no upstream spec equivalent) modeled on real Puppeteer usage patterns — see list below. These run in normal CI, independent of any one-time validation against the real puppeteer-sharp codebase.
  • Keep this file current: update it with learnings as milestones land — API decisions that changed, gotchas found while porting upstream tests, anything a future session would otherwise have to rediscover.
  • README sells it: runnable example, RxJS→ReactiveExtensionsSharp comparison snippet, badges, roadmap/parity status — not a dry stub.

Conventions

  • TFMs: netstandard2.0;net8.0;net10.0 for the main library; test project targets net10.0 only.
  • Nullable reference types enabled (Nullable=enable) — a deliberate divergence from puppeteer-sharp (which has them off); compile-time only, doesn't affect binary consumption.
  • Classic (block-scoped) namespaces, StyleCop-flavored .editorconfig (copied from puppeteer-sharp): no this. qualification, expression-bodied members preferred, braces required, 4-space indent, LF endings.
  • TreatWarningsAsErrors=true via root Directory.Build.props.
  • XML-doc completeness (CS1591, SA160x) is suppressed for now — deferred to the M4 docs pass, not worth blocking TDD cycles on it while the API is still moving.
  • Test framework: NUnit + NUnit.Analyzers + NUnit3TestAdapter + NSubstitute (matches puppeteer-sharp's test stack).
  • Layout is flat (src/ReactiveExtensionsSharp/, test/ReactiveExtensionsSharp.Tests/) at repo root, not nested under lib/ like puppeteer-sharp — no samples/demos/docfx yet to justify that nesting. Revisit if the project grows to match that scope.
  • One file per operator under src/ReactiveExtensionsSharp/Operators/ (mirrors rxjs's own layout; also satisfies StyleCop's one-type-per-file rule for free).

Core design

  • Observable<T> implements BCL IObservable<T> for free interop, but the actual reactive engine (Subscription, Subscriber<T>, the Operate() helper) is custom, layered on top — same shape as rxjs's own internals.
  • rxjs's lift()/hasLift virtual-dispatch cascade is dropped for v1 — it exists mainly to support ConnectableObservable/Subject-overridden operator behavior, out of scope initially.
  • Subject<T>/ReplaySubject<T> implement IObservable<T>, IObserver<T> directly (not by subclassing Observable<T>), each exposing .AsObservable().
  • Unit struct stands in for rxjs's Observable<never> cases (cancellation/timeout streams).
  • No hand-written 9-arity pipe(). Primary idiom is extension-method chaining (source.Map(...).Filter(...).Take(5)).
  • IScheduler seam introduced early (one TaskPoolScheduler impl) threaded through time-based operators, so a VirtualTimeScheduler can be added later without rewriting those operators.
  • Observable.FromEvent wraps a .NET event add/remove-handler pair (the Rx.NET FromEvent<TDelegate, TEventArgs> idiom), not rxjs's DOM/EventEmitter-duck-typed overloads — .NET events are a fundamentally different shape, so this is a deliberate deviation rather than a literal port. Doubles as the C# analogue of Puppeteer's own fromEmitterEvent helper.
  • MergeMap/ConcatMap/SwitchMap don't implement rxjs's concurrency-limit parameter (mergeMap(project, concurrent)) — unbounded merge only for now, since Puppeteer's own usage doesn't need it. Add it later if a real use case needs it, rather than building it speculatively.

Puppeteer-essential surface (priority within the broader port)

Puppeteer funnels all its rxjs usage through one wrapper file re-exporting ~40 symbols. These get priority:

  • Core: Observable, ReplaySubject
  • Creation: of, from, defer, fromEvent, timer, concat, merge, race, raceWith, zip, forkJoin, combineLatest, EMPTY, NEVER, identity, noop
  • Async bridge: firstValueFrom, lastValueFrom
  • Operators: map, filter, mergeMap, concatMap, switchMap, mergeScan, bufferCount, take, first, takeUntil, startWith, distinctUntilChanged, defaultIfEmpty, throwIfEmpty, ignoreElements, debounceTime, delay, delayWhen, retry, catchError, tap
  • Custom combinators to replicate in ReactiveExtensionsSharp.Extras: fromEmitterEvent (done, as Observable.FromEvent — see Core design), fromAbortSignal → FromCancellationToken (done), timeout (done), filterAsync (done), retryAndRaceWithSignalAndTimer (done — retry + raceWith(fromCancellationToken, timeout), the combinator behind Puppeteer's Locator actions)

The entire Puppeteer-essential list is now implemented. Zip/ForkJoin/CombineLatest are same-type-array-only for now (see Core design) rather than rxjs's heterogeneously-typed tuple overloads. M2's remaining work is optional breadth beyond this list, toward the "broad RxJS parity" ambition.

Formerly-low-priority items now done across M4/M5/M7: debounce/audit/sample/throttle families, groupBy, the entire window*/buffer* families (windowCount, windowTime, windowWhen, window, bufferTime, bufferWhen, buffer), a full VirtualTimeScheduler + marble/TestScheduler testing utilities, BehaviorSubject/AsyncSubject, share/shareReplay, bindCallback/bindNodeCallback, partition, using, onErrorResumeNext, plus a large swath of the remaining rxjs surface (see M5/M7 below). Still outstanding: a full Scheduler/Action class hierarchy beyond TaskPoolScheduler/VirtualTimeScheduler (optional — the current seam already supports everything built so far), and standalone connectable/publish/refCount as their own API surface (optional — Share/ShareReplay already cover the common case, and ConnectableObservable was explicitly deprioritized back in M0/M1's Core design).

Milestones

See the original plan for full detail. Summary: M0 scaffolding (done) → M1 core engine + first operator slice (done) → M2 broad operator/creation-function expansion + ReactiveExtensionsSharp.Extras + PuppeteerScenarios tests (done: the entire Puppeteer-essential surface is implemented — scheduler seam, all creation functions, all listed operators, ReactiveExtensionsSharp.Extras, and the end-to-end RetryAndRaceWithSignalAndTimer test) → M3 playground validation against a throwaway branch of puppeteer-sharp (done, see below) → M4 long tail (schedulers, marble testing, docs/CI polish, remaining full-parity operators beyond the essential list).

M3 result: a git worktree at /Users/dario/Code/hardkoded/puppeteer-sharp-rx-playground (branch experiment/rxjs-sharp-playground, off origin/master, local-only/unpushed) with a throwaway lib/PuppeteerSharp.RxPlayground.Tests/ project (not in PuppeteerSharp.sln, not wired into CI). All 4 tests pass against a real launched Chrome (auto-downloaded via BrowserFetcher in a [SetUpFixture], same pattern puppeteer-sharp's own test suite uses): a Locator-style click helper built entirely from Observable.Defer + RetryAndRaceWithSignalAndTimer retries until a delayed-appearing element exists and clicks it, and correctly times out with TimeoutException when the element never appears; Observable.FromEvent correctly wraps both a real EventHandler<ConsoleEventArgs> event (IPage.Console) and a real plain EventHandler event (IPage.Load). This is real end-to-end proof the port's API works for its intended purpose, not just synthetic unit tests.

GitHub remote is live as of this milestone: https://github.com/hardkoded/ReactiveExtensions-Sharp (public), main pushed and tracked, build.yml running green on every push. Milestones up to this point landed as local feature branches merged into main with --no-ff rather than real PRs — going forward, prefer opening real PRs against origin per the standard workflow rules (fresh branch off latest main, ready-to-tag label) now that a remote exists.

API docs are live at hardkoded.github.io/ReactiveExtensions-Sharp — publish-docs.yml builds the DocFX site and deploys it via GitHub Pages (Pages itself had to be enabled once via gh api --method POST repos/hardkoded/ReactiveExtensions-Sharp/pages -f build_type=workflow, not something the workflow can do for itself; not something an agent can do without that one-time API call already having repo-admin-level gh auth). Tag-triggered (v*) or manually via workflow_dispatch. The job always emits a NETSDK1045 ("does not support targeting .NET 10.0") annotation on the Build docs step — confirmed harmless (tried installing both 8.0.x/10.0.x SDKs and pointing docfx.json's metadata TargetFramework at net8.0 instead; neither changed the annotation, and the published site's member count/content is verified complete regardless) — it's a docfx metadata internal design-time-build quirk unrelated to this workflow's own config, not worth chasing further.

NuGet publishing uses Trusted Publishing (OIDC-based, no long-lived API key stored anywhere) rather than a NUGET_API_KEY secret — publish-nuget.yml requests a short-lived (1-hour) API key from nuget.org via a GitHub Actions OIDC token (permissions: id-token: write + the NuGet/login@v1 action), then pushes with that. One-time setup (not something an agent can do — requires a nuget.org account login): on nuget.org, username → Trusted Publishing → add a policy with Repository Owner hardkoded, Repository ReactiveExtensions-Sharp, Workflow File publish-nuget.yml (filename only, no path), Environment left empty (the workflow doesn't declare one). Also needs a NUGET_USER repo secret (just the nuget.org profile name, not sensitive, but kept as a secret per the docs' own recommendation over hardcoding it in the workflow). Tag-triggered (v*) or manually via workflow_dispatch.

M4 done: parallelized across independent workstreams in isolated worktrees (VirtualTimeScheduler + marble testing; docs/CI polish; grouping + rate-limiting operators; BehaviorSubject/AsyncSubject/Share/ShareReplay — see Learnings for the integration issues this created and how they were resolved, including a real correctness bug found and fixed along the way). Landed: VirtualTimeScheduler, marble-testing (TestScheduler, Recorded<T>, cold/hot observables), full XML docs across the public surface (doc-completeness rules re-enabled, no longer suppressed), DocFX site config, MarkdownSnippets, the NuGet-publish/docs-publish workflows (dormant at the time — both are now live, see below), GroupBy/Debounce/Audit/AuditTime/Sample/SampleTime/Throttle/ThrottleTime/WindowCount, BehaviorSubject/AsyncSubject/Share/ShareReplay. 294 tests passing at the end of M4. M4 fully closed out only once NuGet Trusted Publishing (see below) and GitHub Pages (see above) both actually went live — that took until v0.1.0 shipped, well after M4's own operator/testing work landed.

M5 done: another 4-way parallel push, this time explicitly toward "broad RxJS parity" beyond the Puppeteer-essential list. Landed: Skip/SkipWhile/TakeWhile/TakeLast/Last/Single; Distinct/DistinctUntilKeyChanged/ElementAt/Find/FindIndex/Every/IsEmpty/Count/ToArray/SequenceEqual; Scan/Reduce/Pairwise/Repeat/RetryWhen/Materialize/Dematerialize/WithLatestFrom/ExhaustMap (plus a new core Notification<T> type); Interval/Range/Generate/Iif and the full window*/buffer* families. 567 tests passing on main, pushed to the now-live GitHub remote. This slice is also where the disposal-cascade architectural gap (see Learnings) got reconfirmed independently three more times and gained real regression-test coverage for the operators it was fixed in (Skip, SkipWhile, Distinct, WindowWhen, BufferWhen) — but it is still NOT fixed at the root for the many pass-through operators from M1-M4 (Map, Filter, Tap, StartWith, ...). That general fix remains the single highest-priority piece of remaining work; see the Learnings entry for the exact mechanism.

M6 done: the disposal-cascade fix from M5's closing note, landed in two parts (see Learnings for the full technical story). Part 1: a SubscribeChild helper applied to the 17 "single stable inner subscription" operators. Part 2: parallelized across 4 worktree agents, applied the same fix plus matching Remove bookkeeping to every "per-value/per-cycle inner subscription" operator (MergeMap/ConcatMap/SwitchMap/ExhaustMap/MergeScan, Debounce/Audit/Throttle/Sample/DebounceTime/SampleTime, Window/Buffer/WindowWhen/BufferWhen/etc., Retry/CatchError/GroupBy). Also found and fixed a real double-registration bug in SubscribeChild itself along the way, and investigated (but did not confirm — see Learnings) a reported Subject<T>.AsObservable() leak. 603 tests passing at the end of M6.

M7 done: the remaining "rarely-used creation functions" from the parity gap list, parallelized across 3 worktree agents. Landed: Partition (delegates to two independent Filter subscriptions, matching rxjs's actual partition.ts, verified against the 7.8.2 tag rather than assumed); BindCallback/BindNodeCallback (0-3 arity overloads, AsyncSubject-backed caching so the wrapped function is invoked once and replayed to later subscribers, ported near-literally from rxjs's bindCallbackInternals including its isAsync/isComplete flag dance for synchronous-callback-then-throw edge cases); Using (per-subscription resource + source factory, resource disposed on complete/error/unsubscribe, requires IDisposable rather than rxjs's structural Unsubscribable); OnErrorResumeNext (sequences sources, advancing on error OR complete unlike Concat, deliberately NOT using rxjs's own "resubscribe via a teardown action" trick since that would incorrectly advance on a plain external unsubscribe too — see the XML doc remarks on Observable.OnErrorResumeNext for the full reasoning). 644 tests passing on main. This closes out the entire "broad RxJS parity" list from M2's Puppeteer-essential-surface note except the explicitly-optional Scheduler/Action hierarchy and standalone connectable/publish/refCount.

M8 done: a systematic test-coverage audit (a background agent compared every ReactiveExtensionsSharp test file against its upstream rxjs 7.8.2 spec file) turned up a real punch list — missing operators, thin test coverage on Puppeteer-essential operators, and (unexpectedly) several genuine bugs — closed via 6 parallel worktree agents. 825 tests passing on main at the end. Landed:

  • New operators: Finalize (try/finally equivalent; subscribes the source with the exact downstream subscriber, no wrapper, so it never introduces the disposal-cascade gap in the first place), RepeatWhen (the Repeat-family counterpart to RetryWhen), SkipLast/SkipUntil/Expand/Max/Min/EndWith, and a long tail of alias/higher-order operators: MergeAll/ConcatAll/ExhaustAll/SwitchAll (identity-projected *Map delegations), CombineLatestAll/ZipAll (ToArray().MergeMap(sources => Combine/Zip(sources)), rxjs's own joinAllInternals pattern), MapTo/MergeMapTo/ConcatMapTo/SwitchMapTo (deprecated upstream, still in 7.8.2), ConcatWith/MergeWith/CombineLatestWith/ZipWith/pipeable OnErrorResumeNext (prepend-source sugar over the existing creation functions), SwitchScan (MergeScan + SwitchMap's switch behavior), and TimeInterval/Timestamp (two new core value types, using IScheduler.Now so a TestScheduler can drive them deterministically in tests).
  • Feature completions: Retry/Repeat gained a notifier-selector-based delay overload alongside the existing fixed-TimeSpan one (rxjs's RetryConfig/RepeatConfig split into two overloads rather than one config object), plus Retry's resetOnSuccess flag; Share gained a ShareConfig<T> overload (resetOnError/resetOnComplete/resetOnRefCountZero/custom connector), matching rxjs's defaults exactly for the existing no-config overload's unchanged behavior.
  • Real bugs found and fixed, independent of the coverage gaps being closed (this is the most important outcome of M8 — the audit was meant to find missing tests, but writing them surfaced actual correctness bugs):
    • Concat and Merge (the creation functions, not ConcatMap/MergeMap) had the disposal-cascade bug from M6 — but M6 never touched them, since M6's scope was operator extension methods, and Concat/Merge live in Observable.Creation.cs as creation functions with their own independent subscribe logic. Both subscribed each source via the delegate-based Subscribe(onNext:...) overload, assigned into a SingleAssignmentDisposable/disposable list after Subscribe returned — too late for a fully-synchronous source to observe its own disposal mid-emission, the exact root cause documented at length elsewhere in this file. Fixed with the same SubscribeChild (Merge, since each source has exactly one stable subscription) / direct-Subscriber.Create-with-Remove-on-completion (Concat, since it's a per-cycle sequential resubscribe) patterns already established. Lesson: "operators are fixed" and "creation functions are fixed" are NOT the same claim — audit creation functions separately for this bug class if more surface here.
    • CombineLatest completed early if a source completed without ever emitting a value — verified against the real 7.8.2 combineLatestInit source (not assumed): it just decrements an active counter per source completion with no special case for a source that never emitted, so the fix removed a !hasValue[index] || early-completion condition that didn't match upstream.
    • ConcatMap didn't finalize one inner observable before moving to the next queued value. Subscriber<T>.OnCompleted() runs its callback (which calls SubscribeNext()) before its own finally { Unsubscribe(); } — so without forcing the just-completed inner's teardown eagerly first, it only ran after the next inner had already been created and possibly run synchronously. Fixed by calling Unsubscribe() explicitly right before SubscribeNext() (idempotent, so the later automatic one is a no-op) — matches rxjs's own "finalize before moving on" guarantee.
    • TakeUntil's notifier subscription had no onError handler at all, so a notifier error fell through to RxConfig.OnUnhandledError and crashed the test process outright the first time it was actually exercised (this had simply never been tested before — the operator's own XML doc already claimed notifier errors were ignored, but nothing enforced it). Real rxjs's takeUntil.ts passes an explicit noop as the notifier's error handler for exactly this reason (verified against the 7.8.2 source, not assumed). Fixed by passing an explicit no-op onError, making the existing doc comment's claim actually true. A worse near-miss here: my own task prompt to the agent that found this incorrectly assumed, by false analogy with SkipUntil (which genuinely does propagate notifier errors), that TakeUntil should too — the agent faithfully wrote a test matching that wrong assumption, which is what triggered the crash. Lesson: don't assume two similarly-named operators share a behavior just because one's sibling does; check the real upstream source per-operator, especially when writing the task prompt for a delegated agent that won't independently re-derive it.
  • Two worktree agents (out of six) hit the session's usage limit mid-task on this round and required manual completion (finishing their in-progress work directly rather than restarting the whole task) — both had left substantial, correct progress behind (implementations complete, just missing tests, or vice versa), confirming the pattern already noted for M7's Using/OnErrorResumeNext branch: a mid-task session-limit failure is not the same as a broken task, and the in-progress diff is usually worth inspecting and finishing rather than discarding.

Learnings

  • Operators must catch their own user-callback exceptions, not Subscriber. Ported rxjs's tap-spec tests ("should raise error if next/error/complete handler raises error") revealed that a generic try/catch-and-forward inside Subscriber<T>.OnNext/OnError/OnCompleted double-swallows exceptions across nested operator chains (the innermost Observable<T>.Subscribe's own try/catch intercepts the rethrow before it reaches the outer operator, and the target subscriber is already stopped by then, so the forward becomes a no-op). The correct, rxjs-faithful pattern: Subscriber<T> does plain guarded dispatch with no exception handling of its own; each operator (Map, Filter, Tap, CatchError, ...) wraps its own call into user-supplied delegates (project functions, predicates, side effects) in a local try/catch and explicitly calls subscriber.OnError(ex) on failure. Follow this pattern for every future operator.
  • Unhandled errors (no error handler anywhere in the chain) must not throw synchronously. For the same structural reason above — nested Subscribe try/catches would intercept and swallow a synchronous rethrow before it reaches the caller. RxConfig.OnUnhandledError (default: throws on the thread pool) mirrors rxjs's config.onUnhandledError, which exists for exactly this reason, not just as a nice-to-have.
  • Subscriber<T>.Create(...) cannot be a static method on Subscriber<T> itself — triggers CA1000 (no static members on generic types). Factory lives on a non-generic sibling Subscriber class, same pattern as Observable/Observable<T>.
  • StyleCop.Analyzers 1.1.118 mishandles primary-constructor base-list colons: private sealed class Foo(int x) : IBar triggers a spacing conflict where SA1009 ("closing paren should not be followed by space") and SA1024 ("colon should be preceded by space") fight over the same ) : token pair, and dotnet format cannot converge — it even left literal <<<<<<</>>>>>>> conflict markers in files during a multi-TFM format pass. Fix: don't use primary-constructor syntax on classes that also declare a base/interface list; use a classic constructor + fields instead.
  • Target-typed new()/new () has the same oscillation problem (SA1000 "keyword must be followed by space" vs. the whitespace formatter's own preferred style never agreeing under --verify-no-changes). Avoid target-typed new() — always spell out the type (new Observable<T>(...), new object()), which sidesteps the conflict entirely and reads clearer anyway.
  • AnalysisLevel=latest-recommended + TreatWarningsAsErrors is too strict for a fast-moving greenfield library. Full NetAnalyzers "recommended" severity brings in CA1032 (standard exception ctor overloads), CA1063/CA1816 (full IDisposable(bool) + finalizer + GC.SuppressFinalize ceremony), CA1062 (null-check every public parameter), CA1716 (rename params that collide with keywords in other CLR languages, e.g. error), CA2000 (flags intentional-ownership-transfer disposable patterns as leaks) — all as build errors from day one. These are suppressed project-wide (see both .csproj NoWarn lists) until the M4 polish pass; don't remove the suppressions piecemeal without re-litigating the whole list, since they were added together for this exact reason.
  • Subject<T> needs the same delegate-based Subscribe(onNext, onError, onComplete) convenience overload as Observable<T> — without it, callers can't pass a bare method group/lambda directly to a Subject<T>/ReplaySubject<T> instance (only to .AsObservable()), which real rxjs subjects support directly.
  • ReplaySubject<T>.OnNext must check the inherited stopped/disposed state before buffering, not just before forwarding — otherwise a value nexted after OnComplete/OnError gets buffered (and later wrongly replayed to new subscribers) even though it's correctly not forwarded to current subscribers. Required adding a protected bool IsStopped accessor on Subject<T> for subclasses.
  • SingleAssignmentDisposable is used as a "reassign, don't dispose-the-old-one" cell, not the throw-on-second-set type Rx.NET has under the same name. This is only safe because every place it's reassigned (CatchError, Retry, ConcatMap, ...) reassigns after the previous disposable has already self-torn-down via the normal Subscriber.OnError/OnCompleted → Unsubscribe finally-block path. If a future operator needs to swap to a genuinely-still-live disposable, this type is the wrong tool — reach for an actual serial-disposable-with-dispose-on-reassign instead. (Take and First used to use this type for their source subscription too, but don't anymore — see the resolved synchronous-source bug below for why.)
  • C# has no bottom type, so Observable<Unit> sources that only ever error (FromCancellationToken, Timeout) can't be passed directly to RaceWith<T>(Observable<T>...) the way rxjs passes Observable<never> anywhere thanks to structural typing. The fix used here: .Map<Unit, T>(value => throw new InvalidOperationException("unreachable")) — safe specifically because these sources never call OnNext, only OnError, so the projection genuinely never runs. Don't reuse this trick for a source that might actually emit.
  • Time-based tests must not assert synchronously right after Subscribe() when the operator under test uses TaskPoolSchederler (real, async delays) — Retry's delay, Timer, etc. all complete on a background thread. Every such test needs a ManualResetEventSlim (or similar) that the onComplete/onError callback signals, with the assertions running after signal.Wait(timeout). Got bitten by this once already (a "should retry until success" test that asserted 0ms after subscribing, before any retry had fired) — don't repeat it.
  • IScheduler.Schedule cancellation on TaskPoolScheduler relies on the CancellationTokenSource passed into Task.Delay being cancelled before the delay elapses; Race's loser-unsubscription path exercises this directly (a losing branch's Timer must never fire its callback after the race is won) and is covered by a test asserting the loser's side effect never runs.
  • SwitchMap is the one operator so far where the "SingleAssignmentDisposable never disposes what it's replacing" limitation (see above) actually bites: switching to a new inner observable must tear down the still-live previous inner, unlike Take/First/CatchError/Retry where the thing being replaced has always already self-torn-down by the time of reassignment. Fixed by explicitly calling innerSubscription.Disposable?.Dispose() right before assigning the new one. Caught by a test asserting a value pushed into the abandoned inner after switching never reaches the subscriber — don't skip that kind of test for future switch-like operators (exhaustMap, switchScan, ...).
  • Delay cannot be built on DelayWhen/MergeMap the way rxjs implements it, despite that being the obviously elegant reuse — MergeMap subscribes to one independent Timer per value concurrently, and TaskPoolScheduler's real, thread-pool-backed timers don't guarantee same-duration timers fire back in scheduling order. A synchronous burst (Of(1,2,3).Delay(20ms)) came back as [3,1,2] in a test run. Fixed by giving Delay its own single-queue implementation (only one timer ever active, always for the queue head) — see Operators/Delay.cs for the tradeoff this introduces (delays stack under bursts rather than each item getting a uniform per-arrival offset). Lesson: any future operator built from MergeMap/Race/concurrent-timer composition needs an explicit ordering test, not just a values-are-eventually-correct test — this bug would not have been caught by an Is.EquivalentTo assertion, only by Is.EqualTo checking order.
  • Parallelizing M4 across isolated worktrees off the same base commit means the "remove doc-completeness NoWarn suppressions" branch and every branch that added new files raced each other. The docs-pass branch only added docs to files that existed when it was created — any file added by a sibling branch (Testing/*.cs from the scheduler workstream, GroupBy.cs/Audit.cs/etc. from the grouping workstream) had no docs at all, so once both were merged into main (NoWarn suppressions gone, undocumented new files present), the build broke. Fixed by hand after each merge rather than re-suppressing. Lesson: when parallelizing a "remove a project-wide suppression" task alongside other code-adding tasks, either land the suppression-removal branch dead last, or budget an integration pass to add docs to whatever landed in between — this isn't specific to docs, it'll happen with any project-wide-enforcement change parallelized against feature branches.
  • RESOLVED (Take/First's own case) but BROADER THAN FIRST THOUGHT — systemic gap, partially addressed, not fully closed. rxjs's own "should stop listening to a synchronous observable when unsubscribed" test (a hand-rolled source loops for (i = 0; !subscriber.closed && i < 10; i++) { sideEffects.push(i); subscriber.next(i); }) genuinely failed in ReactiveExtensionsSharp for Take/First alone. Root cause: their internal subscription to the source was sourceSubscription.Disposable = src.Subscribe(onNext: ...) (a SingleAssignmentDisposable wrapping the return value of Subscribe), with sourceSubscription.Dispose() called from inside that same onNext callback once the count was reached — but for a fully-synchronous source, the whole loop runs to completion before Subscribe returns, so sourceSubscription.Disposable is still unassigned at the moment .Dispose() is called; the real inner Subscriber<T> the source loop actually checks .IsDisposed on isn't created/disposed until too late. Contrast with real rxjs: a newly-constructed operator Subscriber registers itself as a child of its destination subscriber immediately at construction (destination.add(this)), before the source is ever subscribed to, so a downstream complete()/unsubscribe() cascades down synchronously even mid-loop. Fix applied to Take/First/ShareCore: build the inner Subscriber<T> directly via Subscriber.Create(...) into a local variable before calling src.Subscribe(...). Then a sharper finding while porting Skip/SkipWhile: the SAME failure reproduces for Filter().Take() — i.e. any plain pass-through operator (Map, Filter, Tap, StartWith, DistinctUntilChanged, ...) sitting between a synchronous source and an early-completing downstream operator (Take, First, TakeWhile, Skip, ...) also fails to cascade the stop, because it never checks whether forwarding a value just caused its own downstream subscriber to become disposed. This is NOT limited to operators that complete on their own — it's about every intermediate link in the chain needing to notice "did forwarding this value dispose whoever I'm forwarding to?" and, if so, immediately tear down its own upstream subscription too. Skip/SkipWhile were given this cooperation explicitly (check subscriber.IsDisposed right after every forward, cascade-dispose the inner subscriber if so — see Skip.cs) since their own ported tests required it. Map/Filter/Tap/StartWith/DistinctUntilChanged/DefaultIfEmpty/ThrowIfEmpty/IgnoreElements were NOT retrofitted with this check — confirmed vulnerable, deliberately left as-is to avoid a sweeping, risky change under time pressure, but this is real debt. The actually-correct general fix (verified by hand-tracing, NOT yet applied broadly — see the risk finding below for why): register a newly-created internal Subscriber<T> as a child of the downstream subscriber via downstreamSubscriber.Add(innerSubscriber) immediately after construction, before calling src.Subscribe(innerSubscriber) — mirroring rxjs's destination.add(this) at construction time. Since Subscriber<T>.OnCompleted/OnError already call Unsubscribe() (= Dispose()) on themselves in a finally block, this one change makes ANY downstream completion/error/disposal automatically cascade up through the whole chain via the ordinary Subscription child-finalizer mechanism — no per-operator "check IsDisposed after every forward" boilerplate needed anywhere. Hand-traced end-to-end for Of(1..10).Map(x=>x).Take(3) with this fix applied to both Map and Take: disposing the terminal subscriber at seen>=3 correctly cascades child-by-child all the way up to the Of/From loop's own IsDisposed check, stopping it after exactly 3 iterations — confirmed correct, not just plausible. The literal ported synchronous-source tests for Take/First/Skip/SkipWhile/Distinct are permanent regression coverage — don't remove them, and add one for any new operator in this same shape. A third independent M5 agent reconfirmed the exact same root cause with Map().Take() (see the set/predicate/aggregation slice) — three separate confirmations now, not a fluke. The risk that split this into two parts. Subscription doesn't implement the "a child removes itself from its parent's finalizer list when it disposes independently" behavior real rxjs has (this was explicitly skipped back in M1 as "not essential" — see the original Subscription-spec.ts port). For an operator with exactly ONE stable inner subscription for its whole lifetime, subscriber.Add(innerSubscriber) once is completely safe — nothing ever needs removing early. But for an operator that creates a NEW inner subscription per value or per cycle and expects the old one to just get garbage collected once it completes on its own (MergeMap, ConcatMap, SwitchMap, ExhaustMap, MergeScan, GroupBy's per-key groups, Audit, Debounce, Sample, Throttle, Window*, Buffer*'s per-cycle closing notifiers, Retry's per-attempt resubscription, CatchError's replacement-on-error), blindly Add-ing each new one without also subscriber.Remove(...)-ing it once it completes would make the downstream subscriber's finalizer list grow unboundedly for a long-running stream — trading the disposal-cascade bug for a memory leak. Part (1) DONE (M6). Added OperatorHelper.SubscribeChild<TSource, TResult>(this Observable<TSource> source, Subscriber<TResult> downstream, Action<TSource>? onNext, Action<Exception>? onError, Action? onComplete) in src/ReactiveExtensionsSharp/OperatorFunction.cs — builds the inner Subscriber via Subscriber.Create(...), calls downstream.Add(innerSubscriber) before subscribing, then returns null (see the double-add finding right below for why it's null and not the inner subscriber). Its doc comment repeats the "only for a single stable inner subscription" caveat so it can't be reached for by future per-cycle operators without reading why. Applied to all 17 "single stable inner subscription" operators: Map, Filter, Tap, StartWith, DistinctUntilChanged, DefaultIfEmpty, ThrowIfEmpty, IgnoreElements, TakeUntil, TakeLast, Count, ToArray, Materialize, Pairwise, Scan, Reduce, WithLatestFrom — each just swaps src.Subscribe(...) for src.SubscribeChild(subscriber, ...). Regression coverage in test/ReactiveExtensionsSharp.Tests/Operators/DisposalCascadeTests.cs (synchronous self-checking source composed with Take, through Map/Filter/Tap/Scan/DistinctUntilChanged, plus a multi-operator chain). Part (2) DONE (M6), parallelized across 4 worktree agents. Every operator on the "per-value/per-cycle inner subscription" list now has both halves of the fix: the child-registration cascade AND the matching Remove once that inner subscription naturally ends. Group 1 (mapping-to-observable): MergeMap/ConcatMap/MergeScan build the per-value inner directly and Remove it onComplete; SwitchMap/ExhaustMap reuse one SingleAssignmentDisposable-style slot for their whole lifetime (added to the downstream once), so switching in a new inner just reassigns the slot — no Remove needed there, only the assign-before-subscribe ordering fix. Group 2 (rate-limiting): Debounce/Audit/Throttle rebuild their per-value duration subscriber and Remove it when that cycle closes; Sample's notifier turned out to be single-lifetime (no Remove needed, direct SubscribeChild); SampleTime/DebounceTime (independent scheduler-based implementations, not built on the non-Time variants) get the same Add/Remove treatment for their per-tick scheduled disposable. Group 3 (windowing): Window/Buffer (single-lifetime boundary/notifier) and WindowCount/BufferCount (no notifier) just needed the outer-subscription fix; WindowTime/BufferTime's per-cycle timers were already correctly Added/Removed pre-M6; WindowWhen/BufferWhen needed both the outer-subscription fix and (new) Add/Remove on the per-cycle closing-notifier subscriber — previously that notifier was never registered with the downstream subscriber at all, only torn down via a bolted-on final Subscription wrapper, which is now gone. Group 4 (retry/replace/group): Retry gets the per-attempt Add/Remove and a previously-missing disposed-check before recursing into the next attempt (a stack-overflow guard, matching Repeat/RetryWhen's existing one); CatchError's pre-error source subscription gets the fix, its post-error replacement needs nothing extra since it already reuses the downstream subscriber object directly; GroupBy's outer subscription uses SubscribeChild directly, and its per-key duration-notifier subscriber gets Add-before-subscribe + Remove in CloseGroup() — the trickiest case since group cardinality (and therefore notifier count) can be unbounded. 602 tests green after all four merges (up from 573). A new, distinct bug found while wiring this up: SubscribeChild was double-registering its own inner subscriber. Since it already calls downstream.Add(innerSubscriber) internally, every call site written as return src.SubscribeChild(...) (the tail expression of an Operate callback) let that same object flow back up through Observable<T>.Subscribe, which also adds whatever the callback returns as a teardown (if (teardown is not null) { ... subscriber.Add(teardown); }) — registering the identical disposable twice in the downstream's finalizer list. Harmless in terms of behavior (disposing a Subscription twice is a no-op) but a permanent duplicate entry for the operator's whole lifetime, discovered only because the GroupBy agent worked around it empirically (calling SubscribeChild as a bare statement and returning null explicitly, rather than returning its result) without fully diagnosing why that was necessary. Root-caused and fixed at the source: SubscribeChild now always returns null instead of the inner subscriber, which silently fixes all ~25 call sites without touching any of them. Regression test: DisposalCascadeTests.ShouldNotRegisterTheInnerSubscriptionTwice inspects Subscription's private _finalizers list via reflection and asserts exactly one entry — confirmed (by temporarily reverting the fix) that it fails with a count of 2 beforehand. Lesson: when a helper both performs a side effect (registering a child) AND returns a value that a caller might propagate somewhere that performs the same side effect again, that's a real double-registration risk worth tracing through, not dismissing as "should be harmless." Idempotent Dispose() papered over the actual bug here for a while. A reported Subject<T>.AsObservable() leak investigated and NOT confirmed — the code was already correct. The window/buffer M6 agent flagged AsObservable() => new Observable<T>(subscriber => Subscribe(subscriber)) as silently picking Observable<T>'s Action<Subscriber<T>> constructor overload (which discards whatever the delegate returns) instead of the Func<Subscriber<T>, IDisposable?> one — which would drop the RemovalDisposable that Subscribe returns and leak every AsObservable() subscriber forever. Investigated by hand-reverting to a byte-for-byte copy of the pre-fix line, deleting obj/bin to force a clean rebuild (ruling out incremental-build staleness), and running a reflection-based regression test (SubjectTests.AsObservable_DisposingTheSubscriptionShouldRemoveTheObserverFromTheSubject, inspecting Subject<T>'s private _observers list) against that exact original code — it passed. C# actually resolves the tie correctly: when a lambda with an expression body is compatible with both an Action<T> overload and a Func<T, TResult> overload of the same arity, the language prefers the Func one (the "natural type" with a return value) over silently discarding the return for Action — so AsObservable() was never broken. Added a one-line comment on AsObservable() calling this out explicitly (so a future reader doesn't have to re-derive it) plus the regression test as a permanent guard against a future refactor accidentally landing on the Action overload. Lesson: a background agent's structural/type-system argument for a bug can be wrong even when it sounds plausible and cites real language ambiguity rules — verify with an actual clean-rebuilt failing test before recording something as a confirmed bug in this file, not just before recording something as fixed.
  • A fourth, distinct variant of the same underlying class of bug: WindowWhen/BufferWhen's closing-notifier logic re-subscribes to a fresh notifier every time a window/buffer closes. If that notifier emits and then completes synchronously (e.g. Observable.Of(x)), a naive SingleAssignmentDisposable reassignment reopens the window/buffer twice for one closing event (once on the notifier's next, again on its immediately-following complete) — a recursive re-subscription hazard, not an early-unsubscribe one, but fixed with the same underlying technique (build the Subscriber<T> directly and assign it to its slot before the notifier's Subscribe call can fire synchronously). Caught by a regression test asserting the closing-selector factory is invoked exactly the expected number of times (not double). If you write another operator with a "resubscribe to a fresh per-cycle notifier" shape, check for this specifically — it's easy to miss since the symptom (a stack overflow from infinite reopening, or a silently-doubled count) doesn't look like the classic early-unsubscribe symptom.
  • Generate's 4-overload rxjs surface doesn't fully translate to C#: the "no condition, with resultSelector" combination was dropped because, with generic Func<...> parameters of matching arity, C# overload resolution can't disambiguate it from the "no resultSelector" overload when a throw-expression lambda is involved (its inferred return type is ambiguous) — hit real CS0419/CS0121 errors trying to port rxjs's own condition/iterate-throws tests literally. Callers needing that specific combination pass condition: _ => true explicitly instead. A reminder that not every rxjs overload set has a clean C# equivalent — don't force it when the type system genuinely can't distinguish the cases; document the gap and move on (same spirit as the Zip/ForkJoin/CombineLatest same-type-array-only decision).
  • Pairwise reentrancy bug, found and fixed while porting rxjs's own "recursively re-enterable" test: it updated its previous state slot before forwarding the pair downstream instead of after — so if the downstream subscriber reacted to a value by synchronously pushing a NEW value back into the (shared, e.g. Subject-backed) source from within that same call, the reentrant emission would see an already-updated previous, producing a wrong pair. Fixed by capturing/emitting first, updating previous last, matching rxjs's own pairwise.ts ordering exactly. General lesson: any operator that carries mutable "previous value" state and calls into a user-supplied downstream synchronously must update that state AFTER forwarding, not before — same reentrancy-ordering discipline as DistinctUntilChanged already follows (check that operator's code for the same "update state, then emit" vs. "emit, then update" distinction before assuming either order is safe by default).
  • Repeat/RetryWhen needed the Take/First direct-Subscriber.Create pattern too, but for a different reason than early-completion: naively mirroring Retry.cs's recursive-resubscribe structure has two failure modes for a fully-synchronous, repeatedly-completing source — (a) no check for the downstream already being disposed before recursing into the next cycle, a real StackOverflowException risk once count is large/infinite and downstream is something like Take; (b) rxjs's own ported "finalize before starting the next cycle" test can't hold if the next cycle's resubscription happens inside the still-executing Subscribe call for the current cycle, before that cycle's own teardown had a chance to run. Both fixed with the same direct-Subscriber.Create-then-explicit-check-and-Dispose technique used for Take/First, but applied to an "accumulate then recurse" shape rather than a "complete early" one — this pattern is broader than the name "early-unsubscribe fix" suggests; think of it generically as "always hold a live reference to your own inner subscriber so you can inspect/dispose it synchronously, never rely on Subscribe's return value being assigned in time."
  • Repeat(0)/Repeat(-1) return EMPTY without ever subscribing to the source — Retry(0)/Retry(-1) still subscribe once. An easy-to-miss asymmetry between two structurally similar operators; matches rxjs's own actual behavior for each (verified against the spec files, not assumed), so don't "fix" one to match the other.
  • Notification<T> (new core type, src/ReactiveExtensionsSharp/Notification.cs, main ReactiveExtensionsSharp namespace) backs Materialize/Dematerialize — deliberately NOT a reuse of ReactiveExtensionsSharp.Testing.Recorded<T>, despite the superficial similarity (both tag next/error/complete). Recorded<T> carries a virtual-clock timestamp that has no meaning for a core operator, and core operators shouldn't depend on the ReactiveExtensionsSharp.Testing namespace, which exists for tests. Same CA1000 static-factory-on-a-non-generic-sibling pattern as Recorded/Subscriber.
  • WithLatestFrom subscribes to the "other" observables before the primary source, matching real rxjs's own subscription order — required for the case where an "other" observable emits synchronously during its own subscribe call, which must be visible by the time the primary source starts emitting. If you add a similar "combine primary + others" operator, check subscription order against the actual rxjs source rather than assuming primary-first is safe.