I’ve just released Rihanna as an initial implementation of the concepts discussed here. It uses a very simple single dispatcher with a poll-based approach, which nonetheless has a pretty high throughput of around 1.5k jobs/s due to it’s use of advisory locks.

The supervision subtree looks like this:

          Supervisor
    (one_for_one supervisor)
              |
              |
JobDispatcher (polls and fires off jobs)    
          (GenServer)       
              /\
             /  \
        Task1 ... TaskN

It also has a GUI that can be used to see work in progress and retry failed jobs:

This first version is well-tested and stable at high concurrency levels and should be good enough for most use-cases of a simple, durable, database-backed distributed queueing system.

I believe that significantly higher performance with less load on the DB is possible by taking advantage of pg_notify and communication between nodes of an Erlang cluster to avoid needless contention on the DB, but this is not ready for release yet.

Thoughts/feedback/questions/bug reports are welcome!

This is how I usually end up doing it. Based on your description, I’d have a GenServer dispatcher process, which is responsible for the lifecycle of jobs. I’d implement polling in a separate process (possibly also a Task). That way, a crash when reading from the database would not affect the dispatcher, and therefore would not cause currently running jobs to fail.

Instead of having a separate Task supervisor, I tend to use the dispatcher as the parent of the job processes. I setup the dispatcher to trap exits, and start each new task with Task.async. With such setup, the dispatcher process receives all the necessary events (task result, task failure) in form of messages.

Using a separate Task.Supervisor should also work. I usually don’t do that, since the supervisor wouldn’t really lift a lot of responsibilities from the dispatcher. The dispatcher would have to setup a monitor to each task, and respond to :DOWN messages, and keep monitors and running tasks in its state. So it doesn’t seem that a dedicated supervisor would bring anything useful to the table.

The supervision subtree could look something like:

                Queue
        (rest_for_one supervisor)
                  /\
                 /  \
Queue.Dispatcher      Queue.DbPoller
  (GenServer)       (Task or GenServer)
      /\
     /  \
Task1 ... TaskN

Sounds costly, why not use a PostgreSQL Stream and get results streamed back to you in real-time whenever a table gets a new entry?

You could then just toss it into GenStage/Flow and let it all handle the concurrency as well.

For note, the PostgreSQL Library that Ecto uses (and thus you’d already have it if you use Ecto) has the Stream calls wrapped up in a nice API for you to use.
/me hopes Ecto gets such interfaces too so we get the auto-conversions and such

Yeah Ecto is quite lightweight and does a lot of caching for you and so forth, really useful to have.

Yep! This is it!

What is the reason you are building a postgres-backed queue?

Because:

1. In systems in which it is not a problem that tasks disappear when the system does a full-system restart, you don’t need any persistence.
2. If you do need persistent tasks, you could use (D)ETS or Mnesia to implement this, without needing the external Postgres dependency.

I believe there already are quite a few packages doing either of these:

• transient background jobs: the built-in Task module or ex_job,
• transient scheduled tasks: quantum,
• persistent backround tasks using mnesia: que,
• persistent background tasks using redis: toniq, exq,
• persistent background tasks using RabbitMQ: task_bunny, roger
• persistent background tasks using Postgres: backy (note: still in early development)

Correct. And I would guess that a large majority of Elixir deployments in the wild are also backed by a Postgres DB.

Sure. But mnesia requires your nodes to have state, and it is a PITA to manage (I have used it in production for a while).

It’s not that heavyweight, but it doesn’t make sense for this application for several reasons:

1. Not everybody likes or wants to use it, they shouldn’t be forced to bring it in for a dependency like this
2. This queue only works with Postgres anyway so the database abstraction layer is moot
3. Using the Postgres driver directly gives you better control over how you manage connections

Sure, I looked into Postgres pubsub and it seems interesting, but adds a little extra complexity. It’s not clear-cut whether it would be more performant than a poll-based approach. My current polling implementation is simple and fast enough, once that is stable in production I’ll look into pubsub.

That supervision tree looks interesting… but why separate the Dispatcher and the Poller? Couldn’t you have the dispatcher set a timer to poll itself and fetch/dispatch the jobs synchronously itself inside the handle_info function?

This library aims to solve a slightly different problem than ecto_job. Note that ecto_job uses SELECT FOR UPDATE and holds transactions open for the duration of the job, this limits your concurrency to one job per connection. For some use-cases this is fine, but I want this to be more of a general purpose job queue.

Polling is surprisingly very performant. The problem with streaming events is lock contention since every Dispatcher will try to get a lock at the same time. While I think you could make this work with advisory locks which are very lightweight, you’d always need a polling system as a failsafe so I’d prefer to get that really rock solid first then consider Postgres pubsub as a potential optimisation.

In addition, you don’t need GenStage at all with a polling system since it’s purely pull-based - even simpler.

I am using a combination of advisory locking and SELECT FOR UPDATE SKIP LOCKED.

Initial benchmarks show a throughput of about 1.5k jobs per second on my local machine which is plenty fast enough for our needs.