Connectable observable sends events multiple times

I'm using rxpy for a realtime audio processing tool. The tool receives two event streams, one that contains audio chunks and one that contains small text like annotation snippets. The flow looks similar to this:

audio ----> .publish() ----> rec ----> p1 ---> p2 ---> p 3 ---> p4 ---> merge ---> output
                                        \              /               /
                                          ---> p5 ----                /
annotations --------------------------------------------------> p6 ---

Here, the p<x> nodes are processing nodes (implemented in numpy/pytorch) and the rec node is a recorder, that writes its input to disk and otherwise passes it on unchanged. I'm using the .publish() call to support the branching that comes after the rec node.

When I open the audio file written by the rec node, every chunk has been written 3 times, which implies that the rec node received every chunk 3 times. Is this intended behaviour? How can I avoid this? I'm worried that downstream nodes (p1-p5) might receive multiple repetitions of the same chunk as well and therefore might not operate as intended. However, the pipeline as a whole seems to work correctly.

I tried a number of variations:

1. No .publish() call: Events in the first pipeline get stuck right before p3. p4 never receives any events.
2. Introduce a separate step before the rec node that drops events if they have the same md5 sum as the previous event (either using a combination of scan and filter or a filter with a class). This makes the audio file look ok, but the overall pipeline becomes prohibitively slow and is essentially broken.
3. Move the .publish() call to a later stage: The results are essentially the same as 1.

Thank you for your help.

Start with the .publish() call and recorder node described in the report, then reproduce the three-branch flow with the audio and annotation streams. Trace subscriptions and verify whether each chunk is written once, documenting the intended behavior and any relevant regression test entry point found in the repository.