Possible missed emissions for replay subject

[nigelp-vypex]

It might be possible that there will be new emission(s) in the middle of replaying the previous emission(s) when subscribing to a (serialized) replay subject.

It seems like the replay loop only takes up to the last element of the queue at the start of the loop. New emission(s) that are added to the queue while replaying are not being retrieved which results to those emission(s) to be missed.

         auto get_observable() const
         {
             return create_subject_on_subscribe_observable<Type, optimal_disposables_strategy>([state = m_state]<rpp::constraint::observer_of_type<Type> TObs>(TObs&& observer) {
                 const auto locked = state.lock();
                 for (auto&& value : locked->get_actual_values())
                     observer.on_next(std::move(value.value));
                 locked->on_subscribe(std::forward<TObs>(observer));
             });
         }

Attached is a simple project that can replicated the issue (at least with my setup).

rpp_replay_test.zip

[AlexInLog]

Hi @nigelp-vypex, thanks for submitting issue!

I think you are right: when it calls get_actual_values - we obtain snapshot of values at this timepoint, so, yeah, it is possible to miss some values while looping inside... but i don't see any trivial solution to solve this. Key points:

1. we can block obtaining new values to subject while new observer is draining queue, but it can be time-consuming operation (for example, if observer is doing sleep_for(1s) on every emission - we would block subject for a long period of time)
2. we can add internal queue to observer to save missed emissions but in this case we would have same issue when would try to drain this internal queue. moreover - it would see "old" emissions long period of time =C
3. we can subscribe observer to subject before draining whole queue but it means it could see emissions with invalid order: old and actual values could mix..

So, looks like first case is most correct in terms of ordering and etc but can cause unexpected behavior for rest observers listening current subject... have to think..

[nigelp-vypex]

Hi @victimsnino,

Thank you for lookin into this issue.

Instead of using a copy of the queue, is it possible iterate on the "actual" queue and probably use the latest last element as the condition for each iteration? Using the queue size could be an option for the condition but will have an issue once the queue reaches the max size.

The time between the end of the loop and the actual subscribe is another window where an emission could occur which may also result to a missing value. If it is possible to avoid this, that will be better I guess.

[AlexInLog]

Instead of using a copy of the queue, is it possible iterate on the "actual" queue and probably use the latest last element as the condition for each iteration? Using the queue size could be an option for the condition but will have an issue once the queue reaches the max size.

I think it is not possible due to race condition: what if current thread reads while another writes? it sounds like UB. To protect this we can lock mutex, but then it is actually key point №1 from previous reply =C same for reaching max size if we could avoild locking somehow for example via lock-free queue.

The time between the end of the loop and the actual subscribe is another window where an emission could occur which may also result to a missing value. If it is possible to avoid this, that will be better I guess.

Yeah ,you are right, it is also why i think first option is most correct, but also most heavy i believe =C