hubertlepicki
Debouncing and aggregating events across cluster nodes
I have a system that emits events, such as:
"record_updated", id: 6, company_id: 1
"record_updated", id: 2, company_id: 2
"record_updated", id: 4, company_id: 1
These events are currently dispatched through Phoenix PubSub, from nodes in an Elixir cluster, and either through Channels or GraphQL subscriptions (Absinthe) get delivered to clients who react appropriately to these updates.
The problem is the amount of traffic, and number of updates.
I would like to debounce these events and group by company_id and only emit them when the updates stop or a configured time period is reached.
So, if the events happen constantly, we would emit one event per company_id, having a list of IDs, rather than single ID in their payload, once every 10s or so.
If there was just one event, I think I’d like to wait like a second and emit it to the client as soon as possible.
For example, given the above 3 events, I would like to transform them and emit just two:
"records_updated", ids: [2], company_id: 2
"records_updated", ids: [6, 4], company_id: 3
Any hints on libraries/tools I could use to simplify this task? Should I be looking at GenStage/Broadway and implement this from scratch or there are some tools out there that do something similar to my requirements that you can recommend?
Most Liked
entone
GenStage would probably work great for this use-case. GenStage — gen_stage v1.2.1
Broadway is built on top of GenStage, and could simplify a few things for you, but I think it definitely helps to have an understanding of the underlying concepts. Introduction — Broadway v1.1.0
hubertlepicki
I don’t think I want to use GenServers and buffer these events in memory. There’s several things i dislike about this approach: from the need to cluster process registration and sort of reconciliation to the fact that this is sort of wrong usage pattern as we’re using processes, which are runtime entities, to partition data, and all the ineffectiveness that this brings.
I am thinking now I’ll go with a PostgreSQL or Redis-based solution where I will in atomic way write the IDs of records that need to be broadcasted to clients as being updated, then have a recurring jobs on each node that will also in atomic way fetch and nullify the IDs and broadcast them to clients over PubSub.
hugobarauna
Maybe there’s something to learn from how Nx.Serving deals with batch size and batch timeout:
https://hexdocs.pm/nx/Nx.Serving.html#module-stateful-process-workflow
But yeah, it looks like something Broadway could help: Broadway — Broadway v1.1.0
Popular in Questions
Other popular topics
Latest on Elixir Forum
Sponsor Spotlight
Courses that'll move you from confusion to "Aha, now I get it!"
Categories:
Sub Categories:
Forums
Our Sponsors
Build Elixir applications with speed and confidence.
Supporting innovation across the BEAM ecosystem.
We build reliable cloud platforms for business-critical systems.
Catch errors, track performance, monitor hosts and more.
Develop your skills with books, videos, and courses.
Producing high quality Elixir screencasts since 2017.
The deployment platform built for Elixir: PaaS ease, VPS control.
Courses that'll move you from confusion to "Aha, now I get it!"