Implement transactional updates

[lorentey]

Given an observable integer a, we can define the observable sum of adding it to itself:

import GlueKit
let a = Variable<Int>(2)

let sum = a + a

The value of sum seems to track the original observable correctly:

sum.value      // => 4
a.value = 1
sum.value      // => 2

However, if we look into the changes reported by sum, we find something surprising: apparently the result of adding an integer to itself can temporarily look like an odd number?!

var reported: [String] = []
let connection = sum.changes.connect { change in reported.append("\(change.old)→\(change.new)") }
a.value = 2
reported.joined(separator: ", ")  // => "2→3, 3→4"
connection.disconnect()

The reason for this is quite simple: as a changes from 1 to 2, the plus operator in a + a receives two changes, for both of its dependencies. It doesn't know these changes aren't independent, so it reports two separate changes to itself, going from 2 to 3, then 3 to 4.

These spurious notifications can cause issues when we build complex observable expressions. At the very least, they mean more changes are generated than strictly necessary, costing performance. But the real problem is that such temporary values aren't valid, and shouldn't be reported at all.

[lorentey]

The way to do this is to update the dependency graph in a breadth-first order; i.e., don't update a dependent observable until all of its dependencies have finished processing their changes.

However, rather than doing this by making the dependency graph between observables explicit, I plan to implement this ordering by complicating the way changes get communicated. I'm switching over to a scheme where observables send not just changes, but change events:

public enum ChangeEvent<Change: ChangeType> {
    case willChange
    case didNotChange
    case didChange(Change)
}
public protocol ObservableType {
    associatedtype Change: ChangeType

    /// The current value of this observable.
    var value: Change.Value { get }

    /// A source that reports changes to the value of this observable.
    var changeEvents: Source<ChangeEvent<Change>> { get }
}

Implementation work is progressing on the transactional-updates branch.

The solution resembles a two-phase commit process. Whenever a Variable needs to change its value,

1. It first sends a hold on a moment, I think a change is coming notification to its subscribers (willChange)
2. Then do whatever is necessary to apply the change to its value
3. If it turns out that no change was actually performed, it sends didNotChange, which means oops, false alarm, I remained the same, sorry about that.
4. If there was an actual change, it sends didChange with the incremental change description.

Derived observables subscribe to these notifications. When they receive a willChange, they forward it to their own subscribers, but then delay sending further change notifications until all such outstanding transactions are closed (by a didNotChange or didChange message). While in this paused state, derived observables will need to keep track of pending unsent changes, and send out a single merged change at the end.

While a transaction is in progress, intermediate values must not appear in the value field of affected observables — so that new connections can be safely set up in the middle of an active transaction. (This is important since observable combinators like ValueMappingForValueField need to be able to react to a change by adjusting their dependencies on the fly, perhaps unsubscribing from/subscribing to observables that are also part of the transaction.)

[lorentey]

So it turns out that consistent reads are way too complicated to implement for collections. However, inconsistent reads (and even inconsistent change notifications) are not the real issue — the real issue is the lack of signaling of transaction boundaries. We may allow observables to temporarily get into inconsistent state, as long as they clearly communicate this, and they let us know when it's safe to retrieve their values again.

The updated plan for transactional updates is to simply send a beginTransaction event at the beginning of a sequence of changes belonging to a single transaction, and send an endTransaction at the end. We need not combine partial changes into a single atomic change, although it is possible (and easy!) to do so using an event transformation.

So the changes source of the sum observable in the example above gets replaced by an updates source, which will report the following updates:

beginTransaction
change from 2 to 3
change from 3 to 4
endTransaction

Of course, the original changes source can be emulated by a stateful source transformation that simply accumulates partial changes until it gets an endTransaction message:

extension ObservableType {
    /// A source that reports changes to the value of this observable.
    /// Changes reported correspond to complete transactions in `self.updates`.
    public var changes: Source<Change> {
        return Source { sink in
            var merged: Change? = nil
            return self.updates.connect { event in
                switch event {
                case .beginTransaction:
                    assert(merged == nil)
                case .change(let change):
                    if merged == nil {
                        merged = change
                    }
                    else {
                        merged!.merge(with: change)
                    }
                case .endTransaction:
                    if let change = merged {
                        sink.receive(change)
                        merged = nil
                    }
                }
            }
        }
    }
}

This mechanism is now (partially) implemented on the transactional-updates branch.

[lorentey]

Fixed by PR #4