Assert onError cuts ahead of onNext

... and doesn't allow intermittent onNext via a race of draining the queue and emitting (the draining thread must be where the skipping happens, not the producer thread).