Make the delivery queue deliver in order, and only serialize

[dwcullop]

Two things, both about SharedDeliveryQueue doing one job properly.

1. It does not deliver in the order it received

Jake spotted this while looking at the DQ design and he is right, though the mechanism turned out a little different from the first diagnosis.

Pending notifications are partitioned into a typed sub-queue per source, and DrainPending picked the next sub-queue with Bitset.FindHighest(). That selects by sub-queue index, not by arrival, so two sources come out reordered. With receipt order A1, A2, Bx, delivery is A1, Bx, A2.

This is reachable through the operators, not just by driving the queue directly. SynchronizeSafe(queue) creates a sub-queue per subscription, so in Page the page requests and the data changes land in different sub-queues and both mutate the same Paginator.

A second queue now records which source each pending notification came from, in arrival order, and the drain loop follows that instead of scanning a bitset.

The sub-queues stay typed, which matters. The obvious alternative is one queue of type-erased notifications, but Notification<T> is a readonly struct in a Queue<Notification<T>>, so queuing one currently allocates nothing, and type-erasing would add an allocation per notification. Keeping payloads where they are and recording only the order avoids that: the order entries are sub-queue references that already exist.

Highest-index-first was deliberate. The comment on DrainPending explains it: newer sub-queues are children of older ones, and a child had to drain before a parent's delivery could dispose it. That requirement disappears with arrival ordering. A disposed sub-queue drops its pending notifications, and any order entries it left behind are skipped when the drain reaches them.

Bitset had no other caller, so it goes, along with the sub-queue index bookkeeping and the compaction pass that kept the bitset dense.

2. Batching does not belong in it

The queue also took an optional callback that fired once per drain cycle. CacheParentSubscription used it as its only emit point, which put a batching policy inside a class whose job is serialization. It also batched more than intended: DrainPending drains whatever is queued, so work another thread enqueued mid-drain landed in the same emission.

CacheParentSubscription now tracks its own delivery frame. Each notification increments a depth counter and the accumulated changes are emitted when it returns to zero. A child that emits synchronously during parent processing is delivered inline by the queue's reentrant path, so it nests inside the parent's frame rather than emitting separately. One upstream notification plus everything it triggers synchronously still produces one downstream changeset, which the six operators built on this class already relied on.

I checked what happens without that batching, out of curiosity about whether it was load-bearing: 34 tests fail, and the failures are all changeset granularity. ClearingParentEmitsSingleChangeSet produces 15 where it expects 2. OrderOfChangesIsPreserved produces 11 where it expects 2, and that one matters beyond the count, because a single Edit doing a Clear() then an AddOrUpdate() should reach a consumer as one changeset. Split up, the collection is observably emptied, which is a state that never existed upstream.

What does change is that a second thread's work is no longer folded into the same emission. It gets its own frame.

No lock is needed around the depth counter. The queue has already serialized delivery, so only one thread is ever inside those methods, and the queue's lock provides the barrier between drains on different threads.

With that moved out, the callback has no consumers left, so the field, the overload that took it and the invoke site are all gone. SharedDeliveryQueue now only serializes.

Tests

Three added to SharedDeliveryQueueFixture. Two fail on the current implementation, which is the point:

Expected {"int:1","int:2","str:hello"}, but {"int:1","str:hello","int:2"} differs at index 1
Expected {"int:0","str:a","int:2","str:b","int:4"}, but {"int:0","str:a","str:b","int:2","int:4"} differs at index 2

The third pins the disposal behaviour the old highest-index-first ordering was protecting, so it does not quietly regress.

Full solution builds on all seven target frameworks. I ran the suite four times end to end for each half of this and again for the two combined, plus the deadlock and delivery-queue fixtures twelve times with and without, since a single pass does not say much about concurrency coverage.

Independent of #1097, which removes gate-holding Rx combinators at the call sites and does not touch drain ordering. Both of these are in main today either way.