Test/Operator Modernization

[JakenVeina]

So, with most of the outstanding issues and feature requests cleared out, I'm interested in dong some pro-active maintenance work, I.E. seeking out bug and enhancements in the existing codebase.

The way to do this, I figure, is to focus on enhancing test thoroughness, across existing operators. There's a few things we've talked about in passing, over the last couple years, that we'd like to be consistent about:

1. Error propagation
   1.1. Errors in source streams should virtually always propagate downstream, particularly in scenarios where there are multiple source streams, or child streams, involved.
   1.2. Exceptions thrown by delegates should be caught and forwarded downstream.
   1.3. .SubscribeSafe() should be used to ensure that immediate exceptions get caught and forwarded downstream.
2. Completion propagation
   2.1. Downstream completion should, generally, wait for all upstream completions to occur. What this means specifically will vary from operator to operator, but ultimately, a completion should occur when there is no possibility for further "next" notifications to occur.
3. Empty changesets
   3.1. Operators should always forward an initial changeset, even when empty, and even when the operator otherwise suppresses empty changesets.
   3.1.1. That's not to say that operators should publish an initial changeset upon subscription, like source collections do, they should just respond to the first changeset they see by always emitting at least an empty changeset.
   3.2. Operators should generally suppress empty changesets after the initial changeset. Exceptions include operators where an option for this is specified by the consumer, or the operator documents otherwise.
4. Changeset consolidation
   4.1. Operators should generate at most one downstream changeset for each upstream notification.
   4.2. @dwcullop proposed recently that we could even do better than this: that when a notification comes in while another is being processed, and has to wait, it can signal to the first notification to hold its changeset, so the second notification can add on to it. Thoughts?
5. Changeset correctness
   5.1. Basically, use .ValidateChangeSets() downstream of operators being tested. This verifies that key values are always correct, items are never removed before being added, etc.
6. Notification correctness
   6.1. Notification sequencing, I.E. "only one error or completion notification can occur, and those should end the stream." This theoretically shouldn't be an issue, so long as we're using System.Reactive base types, but being explicit helps make us future-proof.
   6.2. Notification synchronization, I.E. notifications coming out of a stream should be serialized. I.E. any operator that consumes multiple source streams needs to be exercised for multi-thread safety, along with use of .ValidateSynchronization().

I think that's everything I've had on my mind. @dwcullop @RolandPheasant I'd like to lake sure we're all in agreement on these points before I really start digging into writing new tests.

[RolandPheasant]

All sound reasonable to me.

One more I'd add is we should get rid of the original sort related operators for cache operators. Then we can remove index and previous index from the change.

[JakenVeina]

I'll avoid those as I go through, then, with the intention that they may removed later.

I don't know if it's appropriate to remove cache sorting entirely, although that is basically the approach I was planning to take in VNext. The question is how much value is lost downstream of a sort, due to the loss of keys. What operators do we have, right now, that leverage indexes in cache changes?

[RolandPheasant]

The new SortAndBind, SortAndVirtualize and SortAndPage do not use the current and previous index values at all.

The index is only used by the original Sort, Virtualise and Page operators, which are now obsolete and will be removed - that's why I wrote the new versions. The only question is when do they go.

[JakenVeina]

Yeah, I'm still very much against that decision. We already have outstanding issues from people who are using the deprecated .Sort() in scenarios where they can't swap to .SortAndBind() or any of the other new ones.

The approach that I had in mind for VNext is to still support standalone .Sort(), but have it switch you from keyed changesets to ordered changesets, and support only a reduced set of operations upon ordered changesets.

If the operators you mentioned are indeed the only cache operators that require indexing, then that sounds like a valid approach.

Alternatively, if we ARE still pursuing VNext as the future of DD, there's an argument to just leave all that support in, as-is, but not carry it forward into VNext. That provides the most flexibility for consumers that are entrenched in those older operators.

[RolandPheasant]

4.2. @dwcullop proposed recently that we could even do better than this: that when a notification comes in while another is being processed, and has to wait, it can signal to the first notification to hold its changeset, so the second notification can add on to it. Thoughts?

Sounds complicated. For single change set operators I think it should be for the producer to handle the batches. However, for the merge / combiner related changesets it's definitely worth seeing whether it's feasible.

[JakenVeina]

After ruminating, I think it's a cool idea, but more trouble than it's worth in the average scenario. Ultimately, we're talking about an optimization to reduce changeset count, so only really useful for high-changeset-volume scenarios. What this COULD be is just an operator, so that consumers who can demonstrate having performance issues due to high-changeset-volume can opt-into the behavior. Which we technically, already have, I think, right? A .Buffer() or .Throttle() operator? We could bake this kind of behavior into that, but wouldn't that also be redundant, within an operator that's already batching at a higher level of granularity?

[JakenVeina]

I'm about to check in the updates for cache .Filter() operators, as part of this modernization, and a thought suddenly occurs to me...

Do we actually WANT the behavior described in 3.1? The more I think about it, the more I think we really don't. That is, we should publish an initial changeset, upon subscription, when the collection is not empty, but there's no need for it if it is. As far as I can recall, there's only ONE reason that we would need to do this, and it's for aggregation operators, but.... aggregation operators could just be reworked to not require an initial changeset. Specifically, we can implement that aggregation operators themselves always publish an initial value, even when they don't receive an initial changeset (in which case, they can simply assume that the upstream is empty, and aggregate based on that).

In the process of updating the .Filter() operators to always pass-through an initial changeset, I actually found a coiuple edge cases where doing so is problematic, such as ObservableCache's usage of .Filter() upon the .Preview event. I also toyed with the idea of changing ObservableCache to always emit an initial changeset, even when empty, and found that it screws up some of the .Bind() operators and adapters, that rely on the first changeset as a trigger for "initializing" the bound collection.

All this in mind, I think it's a MUCH simpler solution to just make aggregation operators not require an initial changeset. I don't know how much complexity is baked into those operators, though, so I'm going to see if I can prove that concept before we commit to it.

Are there any other scenarios I'm not thinking of where publishing an initial empty changeset is useful?

[RolandPheasant]

Do we actually WANT the behavior described in 3.1?

This is subtly mind melting! I once regretted have not transmitted empty changes on connection but every time I think my mind is settled on the topic I have a re-think.

One way of looking at the problem is to ask why an empty notification is required.? I can think of a couple case, but not many.

I think the initial changes are often used for load indication. The problem with doing that is it's reactive and the data could take any amount of time to load, so in this particular use case the consumer would be better off wrapping some monitor around the code that is doing the loading and use that for loading indication. Directly monitoring the load from code which loads also makes it easier to say whether a change set is empty because it is still waiting for data to arrive, or whether the load has completed and there's no data.

The other use case for the initial change sets is as you say aggregation.

Aside from those, can there be any other use cases?

[JakenVeina]

The one I found, as I mentioned briefly, was initialization within bind operators.

Specifically, SortAndBindOptions has the ResetOnFirstTimeLoad option, which drives whether target collections are initialized by a single Reset() call, or a sequence of .Add() calls. What I saw when introducing a new empty initial changeset was that this changeset would "eat" the initialization. Upstream .Sort() operators would label THAT as the InitialLoad changeset, so when the "real" initial load (with actual items) comes through later, those would be processed as .Add() calls, instead of a .Reset() call, which is presumably less efficient in some scenarios.

Again, this seems to me like a problem that is a downstream responsibility, not an upstream one. I.E. the .Bind() and/or Adapter in question should recognize when it's more efficient to use a .Reset(), whether that's based on a changeset being the "first" one to load items, or any other criteria.

[JakenVeina]

So, I think I've sufficiently proved the concept that aggregation can be handled by the aggregation operators. There's actually one single .Accumulate() method that powers ALL of the aggregation operators, except for StdDev, and Avg, and those two have their own methods that are virtually identical to the shared one.

That method is currently....

return source.Scan(
    seed:           seed,
    accumulator:    (state, changes) => changes.Aggregate(
        seed:   state,
        func:   (current, aggregateItem) => aggregateItem.Type == AggregateType.Add
            ? addAction(current, accessor(aggregateItem.Item))
            : removeAction(current, accessor(aggregateItem.Item))));

I managed to rewrite it to...

return Observable.Create<TResult>(downstreamObserver =>
{
    var result = seed;

    var hasInitialized = false;

    var synchronizationGate = new object();

    var subscription = source
        .Synchronize(synchronizationGate)
        .SubscribeSafe(
            onNext:         upstreamChanges =>
            {
                var newResult = result;

                foreach (var change in upstreamChanges)
                {
                    switch (change.Type)
                    {
                        case AggregateType.Add:
                            newResult = addAction.Invoke(newResult, accessor.Invoke(change.Item));
                            break;

                        case AggregateType.Remove:
                            newResult = removeAction.Invoke(newResult, accessor.Invoke(change.Item));
                            break;

                        default:
                            throw new ArgumentOutOfRangeException(nameof(source), $"Invalid {nameof(AggregateType)} value {change.Type} encountered within changeset");
                    }
                }

                if (hasInitialized && !EqualityComparer<TResult>.Default.Equals(result, newResult))
                    downstreamObserver.OnNext(newResult);

                result = newResult;
            },
            onError:        downstreamObserver.OnError,
            onCompleted:    downstreamObserver.OnCompleted);

    lock (synchronizationGate)
    {
        downstreamObserver.OnNext(result);
        hasInitialized = true;
    }
            
    return subscription;
});

This version successfully does the following:

• Guarantees an initial value gets published, on subscription, when the source doesn't publish an initial changeset.
• Doesn't generate duplicate initial values, when the source DOES publish an initial, non-empty, changeset.
• Maintains passing results for all 32 tests in the AggregationTests namespace.