Dynamic worker by queueName

[piccinnigius]

Hi! there is any way to create dynamically a worker queue?
for example by using body from invoke.

The need is to have different queues such as embeddings-tasks-acme1, embeddings-tasks-acme2, etc for concurrency

EdgeWorker.start(handler, {
  queueName: "embeddings-tasks", // <-- here
  maxConcurrent: 5,
  maxPollSeconds: 5,
  pollIntervalMs: 200,
  // how long to wait before retrying a failed job
  retryDelay: 60,
  retryLimit: 5,
});

one option can be to transform queueName to be string | (req) => string | async (req) => string

@jumski how much can be difficult to implement?

[jumski]

Hey again, @piccinnigius!

Short answer: it is hard to implement due to limitations of Edge Runtime (see Starting Workers below)

Can you give more details on your use case?

Do you need more queues to do a fair-processing, so one tenant does not block the queue for other tenants?
Or something else?

In the mean time I want to acknowledge few things.

Worker-Queue relationship

• one worker can process only a single queue
• you can have multiple workers, each processing a different queue
• you can also have multiple workers processing the same queue to increase throughput

😞 The one queue per worker is hard to change

This was a deliberate decision when planning the architecture of the worker:

• Edge Functions are short lived and not very resource-abundant, so processing multiple queues in single worker is contrary to those limitations
• You can have 50 edge functions in a free tier so it is easy to support multiple queues by deploying more workers
• It extremely simplifies the implementation and debugging

Starting Workers

Currently starting a worker requires firing an HTTP request, but you are not guaranteed to start new worker every time you call it via HTTP - this is decided by the Supabase API Gateway (kong/kong).

It is a reverse proxy with built in load balancer and it spawns new Edge Function instances depending on the load (to increase HTTP throughput) or if there is no instance at all for given edge function.

Given worker only needs a single http request per instance it is not possible to reliably start new "instances" by firing http requests against given Edge Function.

Dynamic queues won't work in this setup

Your suggestion to create a queue based on the HTTP request payload would not really work with this system, because you cannot guarantee the request will spawn a new worker (Kong decides when to do that).

💡 Workarounds

Creating more workers for the same queue

The simplest thing you can try is to deploy more functions with workers, all processing the same queue using same handler.
This will increase the throughput but cannot guarantee any fair processing.

Implementing less-unfair processing with multiple queues

Given the limitation of worker processing only a single queue
You can create multiple separate queues and a worker (or more) for each and write a simple function that will try to distribute the messages across all queues evenly

[piccinnigius]

hi @jumski glad to hear you again!

my use case is exactly this one:

Do you need more queues to do a fair-processing, so one tenant does not block the queue for other tenants?

in particular i need to limit per-tenant the workers due to models quota. In my case each tenant has their keys and their endpoints, this is already managed inside the handler, but at queue level idk how to split them. For me it's ok to create more workers (one for each tenant), but how can i do it programmatically?

thanks again for your awesome work and support

[jumski]

Thanks for the kind words @piccinnigius, I'm really happy that you find my tool useful. Feel free to share any feedback or use cases! 🙇

I have few ideas how to solve it, but it is really very use-case specific.
Can you share more details?

• how many tenants you have
• how often you add new ones
• are they all sending at the same rate, or can you identify few "spammers"?

Workarounds to consider (we can discuss further)

Tenant per queue

Easiest option but limited to 50 tenants on free tier - just assign a new queue every time you get a new tenant.
You can deploy the queues and workers upfront and just assign them to new tenants as they arrive.
If you run out of queues, you can start assigning more to single queue.

Recent changes in Supabase made it possible to deploy edge functions programmatically (see https://github.com/orgs/supabase/discussions/33720) and it should be rather straight forward to do that when you get a new tenant, if you want to do it programmatically.
I'm thinking about supporting this in future via the upcoming pgflow CLI, but its not a priority right now.

Doing "fair sending" onto multiple queues running same code

You can create multiple queues each running the same handler and distribute the messages evenly onto all the queues manually. Its not ideal but should be very simple to implement and is the closest to the "fair processing" we have on mind:
if you have 10 queues and want to send 100 messages for a single tenant, you send 10 per each queue. This way none of the workers are completely stuck with a single tenant. This logic can easily be wrapped into a postgres functions which under the hood will send batches to multiple queues.

Having dedicated queues for the "spammers"

Straightforward - put your spammers on dedicated queues so they don't block other tenants and make rest share other queues (or do "fair sending" for them).

Proper solution (future)

I refactored Edge Worker heavily week ago (not released yet, living on main) and all the internals are easily swappable, so it will be relatively easy to create a new Poller with a custom read_with_poll sql function that would do the real fair processing.

It should be relatively easy - you just store a "tenant_key" in the message and use it to group messages by tenants, so when worker polls for 10 messages it will prioritize reading as many unique tenants as possible.