Building a fair multi-tenant queuing system
inngest.com
inngest.com
0) Load - I'm too loaded right now, I'm not processing your message. Maybe another worker will. This might be enqueued back into the low latency queue.
1) Role Limit - You've sent too many messages in too short of a time, I'm dequeueing this into the high latency queue.
2) Cost Limit - You've sent too many expensive messages, I'm dequeing this into the high latency queue.
3). Deadletter- Your messages are failing to be processed. We're going to dequeue into deadletter and maybe we're going stop processing them for for now.
So you have the normal low latency queue, the high latency backlog queue, and the deadletter queue. Ideally you have SLOs around end to end queue times and monitoring/alerting to let you and your senders know their expectations may be not being met (i.e 99% of messages are processed within n MS).
Instead we use a priority queue and adaptive concurrency. The priority queue lets you reschedule failed messages with a capped exponential backoff. That way it doesn't significantly impair the good items in the queue while still giving plenty of time to potentially fix the failing items. Adaptive concurrency increases & decreases the concurrency for that queue based on success & failure of workers on the queues items up to some upper & lower bounds. That way a single queue can't waste time scheduling work that's just going to most likely fail.
We still have a dead letter queue, but it only ends up containing bugs since expiration is considered a normal message lifecycle event. So dead letter items just contain messages that were too broken to either send or expire correctly.
The key take away is that as the number of queues increase it's better to just operate on the concurrency & ordering of the queue than to just keep creating more queues for the same work stream.
I think one of the big tools is Shuffle Sharding. The article talks about standard sharding by itself as not being enough to provide robustness in multitenant queues. But Shuffle Sharding I.E. assigning users to virtual groups of underlying queues and enqueueing to the queue with the smallest size gets you pretty far. It can limit throughput for individual users but implementing some simple work stealing logic on the consumer helps make sure you keep your throughput up.
https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQS...
Is there a simple way to get a very rough kind of fairness between tenants in this case? Doesn't have to be actually fair, just fair enough that no tenant gets fully stuck behind a large block of jobs from another tenant.
The idea behind SKIP LOCKED is that you just get the first free job (potentially ordered by e.g. a priority column). Doing that in a way that switches between tenants doesn't seem straightforward. You could execute the query once per tenant, but that would be rather inefficient for many tenants. Is there a reasonably straightforward way to achieve this?
It’s nice to keep things simple, but how does that work at scale? If I have 1,000 users I don’t think that these standard queueing systems can prevent one tenant from impacting another. It doesn’t look like that’s possible in SQS, either.
On the other hand, is that even necessary? At what scale is this important?
It's not a huge problem in that case because the regular priority system still puts the urgent jobs at the front of the queue. But it does sometimes cause confusion for someone looking at this from a single-tenant perspective because the queue looks stuck. There are other ways to fix the confusion, but it would be nice to have a very rudimentary fairness in this simple queue that would ensure that the queue iterates a bit over the tenants.
I can see it being helpful in front of a shared resource that doesn't have any way to control priority or fairness. My use case would be Elasticsearch. I don't want to block any searches from running if there's capacity, but also I don't want misbehaving clients to monopolise resources, and I want to make sure that all customers/users get to run their searches eventually, though I may prioritise some customers/users over others. And I may need my own internal jobs to run at a lower priority than a user search, but still within a guaranteed window (to avoid timeouts on the client side).
Honestly for my use case this could be useful with just a handful of users - less than 10, even.
Now, mind, I'm not thinking of a dedicated handler for each tenant, rather one that can round robin, or rotate through the tenants to pull items specific to that tenant off the queue. This can be the "fairness" arbiter.
With the Postgres SKIP LOCKED technique, you can be selective as to which messages you pull (you're not limited to the head of the queue, but can dig deep if you want).
So, you have one general purpose "head of the line" handler, and then one (or several) that acts like the fellow at the Post Office "Anyone here just picking up mail?", and pull them out of the queue.
Of course the generic version can just switch modes (every 1m, or 100 messages, go into "fair mode" for a short time), vs having different handlers.
That's the critical question. To that I had that there might be a myriad of articles on how someone used postgres to build a message queue instead of using a dedicated message broker, but there isn't a single article showing any form of benchmarking, performance tests, load test results, and more importantly report the correlation between load and performance degradation and eventual overload and resulting failure modes. This information is of critical importance to be able to design systems and pick which tool is suited for the job.
In that scenario I employed an earliest-deadline-first scheduler table so rescheduling was just re-inserting or updating with a new deadline.
I did something that sounds similar with the skip locked technique for a request queueing system supporting conveyor routing in a DC.
Priority alone wouldn't work because requests in different areas of the conveyor had different lead times until the response was required, sometimes 5 feet and sometimes 300 feet. If I made the shorter lead times all higher priority then they could all supersede an older request for longer sections that is now nearing the divert.
So the requests get submitted with a deadline datetime, and that is the index used to select the next request (top 1, order by deadline), worked pretty well.
- Decrement priority of all tenants jobs by 1 each time their job is executed, or increment other tenants' priorities (more ops but better behavior)
- Maintain a separate tenant priority table and join it + use for ordering when fetching the next job
And so on
The first option here would create an enormous amount of writes for each job fetched and likely slow it down to a crawl if enough jobs are in the queue.
Having too many jobs in queue is a problem on its own that should be addressed. Each tenant should be rate-limited or have a reasonable cap on number of waiting jobs.
SELECT * FROM jobs ORDER BY RANDOM() LIMIT 1 SKIP LOCKED
If tenant A schedules 99 jobs and tenant B schedules 1 job, a "fair" algorithm would pick B's job either first or second, RANDOM() will not.
One queu is your normal queue.
A second queue that is your tenant queue (one record for each tenant that has queued items).
You select from your tenant queue, then select from your normal queue. All the while keeping your tenant invariant (one record for each tenant that has queued items).
---
It's certainly more complex but no one said that fair was easy.
* How well can you estimate job cost? How should overestimates and/or underestimates be treated?
* Where, when and how often does resource contention occur between jobs? Is it from a single resource, or from multiple different resources?
* What constraints or behaviors does the contended resource(s) possess (e.g. rate limits, pipelining, etc.) that would materially impact how you determine which job to run next?
* What guarantees do you wish to make (e.g. execution latency, resource availability, etc.)?
* How is resource usage quota determined? Are quotas updated frequently? Are there any circumstances under which you would like to allow users to exceed their quota (e.g. temporary burst workloads, idle periods of time)?
Yes, since the SKIP LOCKED query is just a special sort of SELECT that skips over locked rows, you can use all the normal SQL tools such as ORDER BY to accomplish what you want.
One way is to make the top row fair by randomizing over the tenant ID, for example
SELECT itemid
FROM queue
ORDER BY
md5(tenant_id || current_timestamp::TEXT),
item_id
FOR UPDATE SKIP LOCKED
LIMIT 1;
Note that this will eventually have performance implications for a deep queue since it forces a scan on the whole table. Adding an index on tenant_id may mitigate that, but as always do your own profiling.
The documentation under https://www.postgresql.org/docs/16/indexes-types.html implies that it can be used to access the sorted values - but does that feed through the query planner?
Ideally the query planner figures this out (see the sibling comment), but if not, the query can likely be written to "encourage" the planner, for example by first selecting the top few values of md5(tenant_id, current_timestamp) within a subquery before lateral joining into the SKIP LOCKED table.
If the concern is only that the queue falls over once it gets large enough, you can also LIMIT to a fixed number of rows before doing the outer SELECT. This gives a weaker fairness bound (tenants that have fewer than O(queue size / selected items) rows are at risk of being ignored) but you get a better guarantee of progress under contention.
CREATE TABLE [dbo].[Nodes] (
[Id] UNIQUEIDENTIFIER NOT NULL,
[NetworkId] UNIQUEIDENTIFIER NOT NULL,
[Locked] INT NOT NULL,
[Acknowledged] INT NOT NULL,
);
CREATE TABLE [dbo].[Messages] (
[Id] UNIQUEIDENTIFIER NOT NULL,
[NetworkId] UNIQUEIDENTIFIER NOT NULL,
[NodeId] UNIQUEIDENTIFIER NOT NULL,
[Payload] NVARCHAR (MAX) NULL,
[InsertedOrder] INT DEFAULT (NEXT VALUE FOR dbo.MessageSequence) NOT NULL,
[TimestampUtc] DATETIME2 (7) DEFAULT (sysutcdatetime()) NOT NULL,
);
Select *
From Nodes With (Rowlock, Holdlock)
Where NetworkId = @networkId
And Id = @nodeId
(edit: simplified sql)
We went ahead and implemented our own little queue in postgres based on that to get rid of a wonky thing we had that built on top of redis. What I like about this is that it's simple, robust, and transactional. So, you get some nice guarantees out of that.
For fairness, you might do stuff with partitioning on the priority column. Simply set up your workers to only hit a certain partition. So you have separate workers for higher and lower priority queue items. If you then dynamically lower the priority of tenants sending you lots of traffic, the system becomes more fair for the others. You could work with some simple moving average here. Just an idea, but something like this might work.
Note that SKIP LOCKED is available in Microsoft SQL Server and MySQL and I believe possibly also in Oracle and IBM DB/2.
To the article, I suspect there will be alot of detail challenged here.
There's talk in here about "creating queues" to handle the fairness problem. I might be misunderstanding what the hard bit is, but with a database backed queue you don't have to create a queue - queues are virtual depending on what messages are in there and what other information is available such as username or tag. if you want to be "fair" around username then round robin on that, or fair around some other bit of data then tag the message when it goes into the queue and when you are processing, round robin on that.
What is the hard bit here, what am I missing?
For instance: Suppose you have a team of six people, and you're being asked to tackle two projects for two clients, each of equal value to the business. One project will take two weeks to complete; another will take four months to complete.
How do you determine a "fair" way to divide up those projects? Do you do one before the other? Do you divide up your team's capacity equally and do them in parallel?
This is the same sort of queuing problem described in the post. So a solution to the computer queuing problem might have interesting implications for the broader problem.
> It enables developers to write declarative step functions...
> With thousands of customers running thousands of step functions containing parallel steps....
> If you're familiar with step functions, you probably already see what's happening here...
I've done basically this sorta manually at a smaller scale. Each step was a Postgres table, like "user order initiated," "order confirmed," "order fulfilled," with automated steps (sometimes cronjobs) in between. Sometimes I wanted to expose that status to users. I could see using Inngest at a larger scale.
SWE technical terms are used very loosely, especially on blog posts like this.
> each step is a code-level transaction backed by its own job in the queue. If the step fails, it retries automatically. Any data returned from the step is automatically captured into the function run's state and injected on each step call.
This is one thing I've seen so many companies spending tons of time implementing themselves, and happens _everywhere_ -- no code apps, finance software, hospitals, anything that deals with ordering system...the list goes on.
Glad I no longer need to write this from scratch!
The query logic take into consideration the tenants so, a big tenant (or misbehaving) can't block small ones.
The periodic job is executed every few second, but it's also enqueued when added new tasks to the db (unless it's already enqueued) or when the query logic detect that there is more jobs available, so the lag is minimal.
At the moment I am doing this with only one type of jobs, but I want to expand it to most of the jobs, and let the "query logic" balance the queues.
The preemptive scheduler would cover almost every need mentioned in the article.
The fairness of the BEAM scheduler is not the same as multi-tenant fairness. I'm aware of lcnt in Erlang that helps with contention, but that will have a hit in throughput like any other locks.
Unless I'm missing something?
If you wanted to make it truly fair you could spin up a GenServer per tenant and you would have a max concurrency per tenant of 1, but still all executing equally in parallel without 1 tenant stealing CPU time from the others. Contention is a non-issue. Fairness is built in.
You get the fault tolerance, isolation and observability as a by product too.
That's just my first reaction on reading it.
It does have some drawbacks though: if you have thousands of mirrored or quorum queues, managing the cluster becomes more cumbersome - it can take a while to replace nodes (rebalancing thousands of queues around). Also, the management tools (the web UI especially) don't perform that well if you have 10,000s of queues and 100,000s of consumers
Should a user be able to jump the queue? What if they're your number one driver of revenue? Or they have a time-sensitive function w/ a deadline? Or if they've produced no load on the system in the past $TIME_UNIT. All of these are not fair, just properties to be designed into a product.
In the case of this queue, where each letter represents a user:
[A, Bx10000, C, D, E]
We're being unfair to C, D, and E. Realistically, while working on B's jobs we should have some mechanism to know that latency for C, D, and E are increasing and that we can start assigning them to workers.
Without that, latency for any step function you run is highly variable and impacted by other users. With multi-tenant fairness, latency is purely driven by how well we auto-scale and your own concurrency capacity.
The post here is about multi-tenant fairness in particular, so the intent is that fairness is viewed from a multi-tenant / PaaS lense.
Says who? What if B pays 10000x what ACDE do combined? What if B pays for that capacity?
What if B was silent for the 10000 seconds prior? And is now bursting their load just like ACDE do at other times?
There is no objective fair.
Perhaps the only objectively unfair I could think of is if you paid for a service and it's not rendered, at all, ever. That's probably unfair. But everything else is a grey spectrum in between and depends on the agreed upon properties.
If B's load is consistently that high compared to the others then presumably the system has enough total capacity that those requests don't impact the latency for everyone else. If B pays for dedicated capacity they should ideally get their own queue.
> What if B was silent for the 10000 seconds prior?
Then B should smooth out their load better.
In this extreme an example, not much is lost for B by prioritizing C, D, E first sometimes anyways.
There might not be one single definition of fairness, but preventing one noisy user from impacting other users is a pretty common way to approach it.
> One user should not be able to block another's work.
A multi-tenant architecture implies over committing resources to achieve better economics. Once upon a time, decades ago, when computers were so expense that companies payed for batch processing instead of owning their own, FIFO was acceptable. (And for a tiny part of the market it still is.) But ever since the rise of time-sharing operating systems, users have come to expect short delays for small tasks. If you let one customer monopolize your capacity because they submitted a mountain of work before a second customer submits a handful, you probably won't have the second customer much longer.
There are many other forms of queueing other than fair, so if you're looking for a more specific use case, there may already be an algorithm for it.
https://en.wikipedia.org/wiki/Fair_queuing https://en.wikipedia.org/wiki/Weighted_fair_queueing https://intronetworks.cs.luc.edu/1/html/queuing.html