It looks like there are a number of improvements that could be made so your stages run smoother.

For instance, in Coxir.Struct you are using ets as an in-memory backing store and have these encode/1 and decode/1 functions. Those are getting used a lot and just change a Map into a list to a tuple and back… It looks like you are doing this so you can store it in an ets table which needs tuples … but … you could just do:

:ets.insert @table, {id, data_as_a_map}

Then you will have equivalent lookups:

[{_id, data}] = :ets.lookup @table, id

No moving data between different data structures constantly. Whether or not that is causing your memory usage (doubtful), it will certainly relieve some pressure on the GC and make your code faster.

In Coxir.Stage.Middle.handle/2 there are many places where you are using for list comprehensions, but discarding the results. for creates a list out of every single result … if you do not use those results, the for is storing them in a list for no good reason. What you probably want instead is Enum.each/2 which just calls the function for each entry in the enumerable, IOW: for the side-effects. List comprehensions are there to build lists, not create side effects.

More plainly: for in Elixir is not equivalent to a for-loop in your typical imperative language. Enum.each/2 is closer to loops that do not change external variables, and Enum.reduce/3 is closer to those that do.

There is also interesting code in Consumer like:

handler
|> apply(:handle_event, [event, state])

where you could just do:

handler.handle_event(event, state)

It’s equivalent and probably easier to read, esp as you can then just do:

case handler.handle_event(event, state) do

which is, at least IMHO, a lot easier to read than the pipeline syntax used there It contains lal the information needed in one line (e.g. what is the subject of the case) rather than having to track back…

I also noticed you have call of String.to_atom/1 on externally supplied data … this is a great way to exhaust the atom table (you only get so many!), run out of entries, and DoS your app as an unhappy side-effect You should really look to replace that approach.

Finally … as to where your binary “leaks” are coming from … I assume it is something like this:

• you get a block of json as a binary from the external service
• Jason parses that into smaller strings for your, but each of those smaller strings are just pointers into the bigger one … and that prevents the GC from removing those binaries until the references to the smaller strings are dropped
• those smaller strings are being put into an ets table, so they stay forever until the entry is deleted from the ets table, which means the binaries stay forever until the entries are deleted in the ets table
• since coxir is updating those entries over time from many different json input binaries, each entry in the ets table(s) is keeping multiple json binaries in memory over their lifespan.

You can get around this by calling :binary.copy/1 on the specific sub-strings you are storing in the ets table to see if that is indeed the issue.

Did you end up solving your problem? Quite an interesting thread.

I work on a server that processes a lot of video and audio in real time as well, one of the first things I did is to write my own data pipeline workflow logic, for that I used only OTP. I see you are using gen_stage, after a quick glance I noticed this library buffers data and I will quote the docs:

Defaults to 10_000 for :producer , :infinity for :producer_consumer

From what you described, it feels like you are not consuming/processing data fast enough and the buffer for producer_consumer keeps growing.

I would stay away from gen_stage, or at least make a very simple version of your server that does not use it, just for the sake of comparison.

You may be able to simulate this by simply changing the demand values to 1 to avoid buffering entirely. If this causes the desired change then the memory usage isn’t due to a leak, it’s just due to a buffer configured for more than you want.

https://hexdocs.pm/gen_stage/GenStage.html#c:init/1-options

This API has changed a little bit since I first used GenStage, but yeah you’d lower your max_demand to 1 on the consumer and producer consumer. The reason is that if you stick with the default it’ll buffer up to MIN_DEMAND which is 5_000 items before actually pushing it to the consumers.

This option will look like:

 {:producer_consumer, number, subscribe_to: [{A, max_demand: 1}]}