Repository navigation
Possible missed emissions for replay subject #703
Description
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).
Activity
AlexInLog commentedon Feb 11, 2025
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: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 commentedon Feb 11, 2025
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 commentedon Feb 12, 2025
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.
Yeah ,you are right, it is also why i think first option is most correct, but also most heavy i believe =C