Racing condition in RxJs event order (with multiple subscriptions)

This is another short article about a "strange issue while working on a project" which took me a while to resolve.
The problem was that the order of events got mixed up in a observable chain, so the first event arrived after the second one (and no there were no http calls or other async operations involved).
As it is hard to explain just in words I created a simplified example application to better show the issue (and help me figure out the root cause).

Example application

Components

The main components and dependencies are:

• TasksService
  • Stores tasks (a task is a string to keep it simple) in a ReplaySubject which is accessible to other components by the tasks$ property.
  • Contains a setTasks method as a means for other components to update the tasks.
• TasksFilterService
  • Stores a filter pattern (again a simple string) to match against the stored tasks.
  • Subscribes to tasks$, filters the received tasks and updates tasks$ in TasksService by calling setTasks$
  • An infinite loop is avoided by using distinctUntilChanged --> so tasks$ is only updated in case the filter pattern does lead to a change in the current list of tasks
• Component
  • Subscribes to tasks$ and tasksFilterPattern$ to derive a areTasksValid$ property. Tasks are considered valid if all of them are matching against the filter pattern. In contrast to what is done in TasksFilterService it only checks the list of tasks but does not perform any update in TasksService.
• Template
  • Subscribes to areTasksValid$ and renders the boolean value using the async pipe.

When looking at the structure some may spot the circular dependency between TasksService and TasksFilterService.
TasksFilterService filters (based on the set filter pattern) and updates the tasks in TasksService whenever tasks$ emit. To avoid an infinite loop the TasksFilterService should only perform the update in case the filtered list is different to the current one. As mentioned above this can be done using distinctUntilChanged.

Implementation

The Component sets the tasks filter pattern on init and updates the tasks after two seconds using setTimeout.
As both the Component as well as the TasksFilterService subscribe to task$ my initial guess was that there are two routes the application can take when setTimeout is triggered:

1. Tasks are first received by the Component:
   1. The Component checks if tasks are valid which is then displayed in the Template (areTasksValid$).
   2. Next TasksFilterService filters and updates the tasks. The second emit of tasks$ does not lead to another update due to distinctUntilChanged.
   3. The Component receives the filtered tasks which are again displayed in the Template.
2. Tasks are first received by the TasksFilterService
   1. Same as step 2 in the first option (TasksFilterService updates the task list)
   2. Component first receives the unfiltered list of tasks and checks for their validity. Next the filtered tasks are received (set in step 1) and again checked against the current filter pattern.
   3. As the complete execution is synchronous the change detection runs once and the result from the last task list processed is displayed in Template.

As it turned out none of these routes are taken. What actually is executed:

1. Tasks are received by the FiltersService, filtered and updated
2. Second emit of tasks$ is ignored in FiltersService (due to not having changed)
3. Component first receives filtered tasks list and then the unfiltered one
4. Template renders the final value

The actual sequence is pretty close to the second guess. The important difference (marked in bold) is that the tasks received by Component are out of order. As shown in the sequence diagram below, the filtered tasks list comes before the unfiltered one. My assumption was that the order of events is guaranteed, so I expected the Component to receive the task lists in the order these have been set in TasksService (the initial list before the filtered one).

This explained the bug I was facing in my project where one of the Components was rendering outdated/old values.
Basically one cannot assume emissions to be in order in case of multiple subscriptions to the same Observable where one of them is again updating the source stream (tasks$ in the example). Of course, the problem only occurs in case processing and updating the streams happens within the same tick of the event loop.

The sequence diagram below shows what has been described above.

In addition there have been log statements added indicating where and when the task list update is processed (triggered by setTimeout).

The result rendered in the Template is false because the final list received is the unfiltered one and therefor not valid for the applied task filter pattern.

Solutions

I could come up with two options to guarantee the order of the task list received by Component:

1. Directly perform the filtering in FiltersService so that only the already filtered task list is emitted.
2. Put the update in TasksFilterService on next tick of the event loop. This can be done by using any of the available asynchronous schedulers (like asapScheduler or requestAnimationFrameScheduler). That way it is guaranteed that the filtered task list is emitted after the initial one. In contrast to the first solution it does lead to both tasks list being emitted and, depending on which async scheduler is chosen, two change detection cycles.

To me the first solution has the advantage it is easier to understand and straight forward. In contrast the second solution has the benefit of clearly separating the concern of storing tasks and filtering.

In my project I opted for the second solution since there was a lot more going on in the services and migrating everything into one would have lead to a huge class and a (more) severe violation of Separation of Concerns.

Both versions are implemented in the example application and can be tested by commenting in the associated component in app.component.html. app-tasks-scheduled represents the scheduled implementation and app-tasks-single-stream the version directly filtering in TasksService.

Thanks for reading.