Fix ObservableChangeSet.Create's task overloads and build EditDiff on them - #1206

Part of #1142.

EditDiff on an IObservable<IEnumerable<TObject>> used the synchronous ObservableChangeSet.Create, whose callback only gets an ISourceCache and so has no way to fail. A source error after subscription was rethrown by Rx on the thread that delivered it.

The ObservableChangeSet.Create overloads whose task returns nothing can fail and complete through that task, so EditDiff now uses one, with ForEachAsync making the source's lifetime the task:

ObservableChangeSet.Create<TObject, TKey>((cache, cancellationToken) => _source.ForEachAsync(items => cache.EditDiff(items, _equalityComparer), cancellationToken), _keySelector);

Those overloads did not behave like their analog, Observable.Create(Func<IObserver<T>, CancellationToken, Task>). They never completed when the task did, and because they awaited inside an async lambda, a task that ended on a thread with a SynchronizationContext (a UI thread) had its outcome delivered later, on the thread pool.

They are now composed from Rx operators:

Observable.Using(() => new SourceCache<TObject, TKey>(keySelector), cache => Observable.FromAsync(cancellationToken => subscribe(cache, cancellationToken), TaskObservation).Publish(loaded => cache.Connect().TakeUntil(loaded)));

• Using owns the collection.
• FromAsync observes the task on ImmediateScheduler, so its outcome arrives on the thread that completes it. Unsubscribing cancels the token.
• TakeUntil ends the changes when the task succeeds, and fails them when it fails. It holds no lock while delivering.
• Publish starts the task only once TakeUntil is subscribed to the changes, so a task that finishes during subscription keeps its changes.

The overloads whose task returns an IDisposable or Action are unchanged. They already sit on Rx's Observable.Create(Func<IObserver<T>, CancellationToken, Task<IDisposable>>), which is their analog and disposes clean-up that arrives after unsubscribing. No public signature changes.

Instructions

copilot-instructions.md is the only instruction file that is always in context, and its operator pattern presented Observable.Create as the way operators are built, while rx.instructions.md says composition comes first. It now carries a short version of that rule: compose with existing operators, use ObservableChangeSet.Create rather than routing around it, and the smells that mean a composition was missed. It also says to read rx.instructions.md before writing any Observable.Create or ObservableChangeSet.Create body.

Tests

Cache/ObservableChangeSetFixture:

• TaskCompletesTheStreamAfterItsChanges
• TaskFailureIsDeliveredOnTheThreadThatFailsIt
• TaskThatFinishesDuringSubscriptionStillDeliversItsChanges
• DisposingTheSubscriptionCancelsTheTask

List/CreationFixtures:

• TaskCompletesTheStreamAfterItsChanges
• TaskFailureIsDeliveredOnTheThreadThatFailsIt
• TaskThatFinishesDuringSubscriptionStillDeliversItsChanges

Six of these fail on main. DisposingTheSubscriptionCancelsTheTask pins behaviour that is kept.

Cache/EditDiffChangeSetFixture:

• ResultFailsWhenSourceFailsAfterSubscription (new) fails on main.
• ResultCompletesIfAndOnlyIfSourceCompletes and ResultFailsIfAndOnlyIfSourceFails subscribed to the source rather than to the result of EditDiff, so they never exercised the operator. Both now do.

Notes

Behavioural changes to public APIs. main is on 10.0-preview.

• ObservableChangeSet.Create with a Func<..., Task> now completes when the task completes, and a failing task's error arrives on the thread that fails it.
• The list Func<ISourceList<T>, CancellationToken, Task> overload now connects before running the task, like the cache one, so changes made during the task arrive as they happen instead of being folded into one changeset.
• EditDiff: an error after subscription now arrives as OnError. A source that emits more than once while being subscribed to now produces a changeset per emission, and changes ahead of a failure during subscription are delivered before the error.