[BUG] Thread safety in ObservableCache.Connect (regression)

[pmg23]

The thread safety of ObservableCache.Connect (previously fixed in #377) is broken again by #472.

It is imperative to hold the lock on _locker while fetching the initial changeset and subscribing to changes. Otherwise any changes made from another thread between the point when, for a new subscription, InternalEx.Return computes the initial state, and later the Concat impl subscribes to _changes, will be lost.

DynamicData/src/DynamicData/Cache/ObservableCache.cs

Lines 106 to 131 in 811b32e

	public IObservable<IChangeSet<TObject, TKey>> Connect(Func<TObject, bool>? predicate = null) =>
	Observable.Create<IChangeSet<TObject, TKey>>(
	observer =>
	{
	var initial = InternalEx.Return(() =>
	{
	// lock getting initial changes and rely on a combination of Concat
	// + _changes being synchronized to produce thread safety (I hope!)
	lock (_locker)
	{
	return (IChangeSet<TObject, TKey>)GetInitialUpdates(predicate);
	}
	});
	var changes = Observable.Defer(() => initial).Concat(_changes);
	if (predicate != null)
	{
	changes = changes.Filter(predicate);
	}
	else
	{
	changes = changes.NotEmpty();
	}
	return changes.SubscribeSafe(observer);
	});

(Observed in 7.4.3, but should affect 7.1.16 onwards.)

[RolandPheasant]

Hopefully the PR will fix it.

[pmg23]

Looks good. Thanks for the fast turnaround.

[github-actions]

This issue has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.