[Bug]: Combining Throttle with AutoRefreshOnObservable throws ArgumentOutOfRangeException in ChangeAwareList

[lucdem]

Describe the bug 🐞

I've spotted two different scenarios where combining AutoRefreshOnObservable with throttle causes Exceptions in code that works normally if the Throttle is omitted.

• Scenario 1: When calling Filter->Throttle->Transform after AutoRefreshOnObservable
• Scenario 2: When using AutoRefreshOnObservable to refresh based on an Observable that isn't empty (already contains a sequence), then calling Throttle->Filter/Transform (either one breaks)

Step to reproduce

Code to reproduce both scenarios:

using DynamicData;
using DynamicData.Binding;
using System.Reactive.Linq;

internal class Program
{
	public class Foo(IObservableCache<Bar, Bar> cache)
	{
		public IObservableCache<Bar, Bar> BarCache { get; } = cache;

		public Guid Id { get; } = Guid.NewGuid();
	}

	public record Bar(int a, int b);

	private static void Main(string[] args)
	{
		Scenario1();
		//Scenario2();

		Thread.Sleep(1000);

		Console.ReadKey();
	}

	private static void Scenario1()
	{
		var fooCache = new SourceCache<Foo, Guid>(x => x.Id);
		var barCache = new SourceCache<Bar, Bar>(x => x);
		var foo = new Foo(barCache);
		fooCache.AddOrUpdate(foo);

		fooCache.Connect().Bind(out var observableCollection).Subscribe();

		observableCollection.ToObservableChangeSet()
			.AutoRefreshOnObservable(x => x.BarCache.Connect())
			.Filter(x => x.BarCache.Count > 0)        // if any of these 3 is removed the code works
			.Throttle(TimeSpan.FromMilliseconds(500)) // if any of these 3 is removed the code works
			.Transform(x => x)                        // if any of these 3 is removed the code works
			.Subscribe(x => Console.WriteLine("changed"));

		barCache.AddOrUpdate(new Bar(1, 2));
		barCache.AddOrUpdate(new Bar(4, 5));
		barCache.AddOrUpdate(new Bar(5, 6));
	}

	private static void Scenario2()
	{
		var fooCache = new SourceCache<Foo, Guid>(x => x.Id);
		var barCache = new SourceCache<Bar, Bar>(x => x);
		var foo = new Foo(barCache);
		fooCache.AddOrUpdate(foo);

		fooCache.Connect().Bind(out var observableCollection).Subscribe();

		barCache.AddOrUpdate(new Bar(1, 2));
		barCache.AddOrUpdate(new Bar(4, 5));
		barCache.AddOrUpdate(new Bar(5, 6));

		observableCollection.ToObservableChangeSet()
			.AutoRefreshOnObservable(x => x.BarCache.Connect())
			.Throttle(TimeSpan.FromMilliseconds(500)) // if any of these 2 is removed the code works
			.Transform(x => x)                        // if any of these 2 is removed the code works
			.Subscribe(x => Console.WriteLine("changed"));
	}
}

Reproduction repository

codeInIssue

Expected behavior

Expected Throttle to not alter the behaviour of other reactive operators

Screenshots 🖼️

No response

IDE

No response

Operating system

Windows 11

Version

No response

Device

No response

DynamicData Version

8.4.1

Additional information ℹ️

dotnet 8

[JakenVeina]

The .Throttle() operator is a System.Reactive operator, not from DynamicData, and it is fundamentally incompatible with DynamicData streams. What you want is .Buffer() (followed by .SelectMany().

The .Throttle() operator actually DROPS notifications during the throttling window, instead of just delaying them all. Per documentation...

Ignores the values from an observable sequence which are followed by another value before due time with the specified source, dueTime and scheduler.

In Scenario1(), without the .Throttle() operator, the change set stream going into .Transform() looks like...

ChangeSet<Foo>. Count=1
        Add. 0 changes
ChangeSet<Foo>. Count=1
        Replace. 0 changes
ChangeSet<Foo>. Count=1
        Replace. 0 changes

With the .Throttle() operator, it looks like...

ChangeSet<Foo>. Count=1
        Replace. 0 changes

The Add change disappears entirely, and you end up asking the .Transform() operator to replace an item that it doesn't know exists. Using .Buffer() gets you the throttling behavior, without throwing away change sets.