For several projects I’ve opted for the even dumber approach, that works out of the box with every ORM/Query DSL framework in every language: using a normal table with SELECT FOR UPDATE SKIP LOCKED
I've done even simpler without locks (as no transaction logic), where I select a row, and then try to update a field about it being taken. If 1 row is affected, it's mine. If 0, someone else did it before me and I select a new row.
I've used this for tasks at big organizations without issue. No need for any special deployments or new infra. Just spin up a few worker threads in your app. Perhaps a thread to reset abandoned tasks. But in three years this never actually happened, as everything was contained in try/catch that would add it back to the queue, and our java app was damn stable.
I guess you update it with the assigned worker id, where the "taken by" field is currently null? Does it mean that workers have persistent identities, something like an index? How do you deal with workers being replaced, scaled down, etc?
Just curious. We maintained a custom background processing system for years but recently replaced it with off the shelf stuff, so I'm really interested in how others are doing similar stuff.
No, just update set taken=1. If it was a change to the row, you updated it. If it wasn't, someone updated before you.
Our tasks were quick enough so that all fetched tasks would always be able to be completed before a scale down / new deploy etc, but we stopped fetching new ones when the signal came so it just finished what it had. I updated above, we did have logic to monitor if a task got taken but never got a finished status, but I can't remember it ever actually reporting on anything.
I would set the taken field to a timestamp. Then you could have a cleanup job that looks for any lingering jobs aged past a reasonable timeout and null out the field.
it wont work with a timestamp because each write will have an affected row of 1 beacuse the writes happen at different times. setting a boolean is static
We have a "status flag" column which is either Available, Locked or Processed (A, L and P), an Updated column with a timestamp of when it was last updated, and a Version counter.
When grabbing a new message it selects "Available or (Locked with Updated timestamp older than configured timeout)". If successful it immediately tries to set the Locked status, Updated timestamp and bumps the Version counter, where the previous values of Status and Version has to match. If the update fails it retries getting a new message.
If the Version counter is too high, it moves the message to the associated dead-letter table, and retries getting a new message.
This isn't for high performance. I tested it and got 1000 messages/sec throughput with handful of producers and consumers against test db instance (limited hardware), which would be plenty for us.
I wrote it to be simple and so we could easily move to something AMPQ'ish like RabbitMQ or Azure Service Bus when needed. Overall quite easy to implement and has served us well so far.
That is the sort of thing that bites you hard when it bites. It might run perfectly for years but that one period of flappy downtime at a third party or slightly misconfigured DNS will bite you hard.
But compared to our rabbit setup where I work now, it was dead stable. No losing tasks or extra engineering effort on maintaining yet another piece of tech. Our rabbit cluster acting up has led to multiple disasters lately.
Agreed, I've had my own rabbit nightmares. But setting up a more robust queue on postgresql is easy, so you can easily gain a lot more guarantees without more complexity.
I've done this successfully with a web service front that retrieves jobs to send to workers for processing, by using a SQL table queue. That web service ran without a hitch for a long time, serving about 10 to 50 job consumers for fast and highly concurrent queues.
My approach was:
- Accept the inbound call
- Generate a 20 character random string (used as a signature)
- Execute a sql query that selects the oldest job without a signature and write the signature, return the primary key of the job that was updated.
- If it errors for any reason, loop back and attempt again, but only 10 times, as some underlying issue exists (10 collisions is statistically improbable for my use case)
- Read the primary key returned by that sql query and read it, comparing it's signature to my random one.
- If a hit, return the job to the caller
- If a miss, loop back and start again, incrementing attempts by 1.
The caller has to handle the possibility that a call to this web service won't return anything, either due to no jobs existing, or the collision/error threshold being reached.
In either case, the caller backs for it's configured time, then calls again.
Callers are usually in 'while true' loops, only existing if they get an external signal to close or an uncontrolled crash.
If you take this approach, you will have a function or a web service that converts the SQL table into a job queue service. When you do that, you can build metrics on the amount of collisions you get while trying to pull and assign jobs to workers.
I had inbuilt processes that would sweep through jobs that were assigned (had a job signature) and weren't marked as complete, it actioned those to handle the condition of a crashed worker.
There are many many other services the proper job queues offer, but that usually means more dependencies, and code libraries / containers, so just build in the functionality you need.
If it is accurate, fast enough, and stable, you've got the best solution for you.
Agenda uses this, and we found the hard way on mongo 4 that it can lead to mongo spinning the CPU at 100% if it gets too many at once. No idea if they've fixed it in later versions.
I recently got introduced to this system at work, and also built a new job using it. It works fine, but since I had to implement work stealing to deal with abandoned jobs in a timely manner, I wouldn't dare to use it for actions that absolutely must not happen twice.
Exactly-once is only meaningfully possible if you have a rollback for tasks of unknown completion state - for example if the task involves manipulating the same database as the one controlling the task execution. Otherwise, it becomes the (impossible to solve) two-generals problem between updating the task status and performing the task.
There is actually another possibility: there must be a way to check whether the receiving system has received the message. But this only works if there are no "rogue" senders.
The reason why you want to use skip locked is so that Postgres can automatically skip rows that are being concurrently accessed for updating the "status". You are right, if you update a "status" field you don't really need to worry about advisory locks and skipping rows that are locked but it still helps with performance if you have a decent amount of concurrent consumers polling the table.
PSA: This is a read-modify-write pattern, thus it is not safe under concurrency unless a transaction isolation level of SERIALIZABLE is specified, or some locking mechanism is used (select for update etc).
This should be safe under SI (other than the ABA issue, which isn't even fixed with serializable). The update forces a W-W conflict, which is sufficient to make the behavior serializable under SI (and therefore, I think but am not sure, PG's RR level too).
The part about checking the number of affected rows hints at using `UPDATE ... WHERE ...` which should act as an atomic CAS regardless of isolation level.
Edit: To clarify, I mean `SELECT id WHERE used = 0` followed by `UPDATE ... SET used = 1 WHERE id = ... AND used = 0`
I don't get it :(. Why could the same task be executed more than once? From my understanding, if the UPDATE is atomic, only one worker will be able to set `used = 1`. If the update statement is not successful (affected != 1), then the worker should drop the task and do another select.
With a transaction isolation level below SERIALIZABLE you can have two transactions that both read the old row (with `used = 0`) at the time they perform the update (but before they commit the transaction). In that case, both transactions will have performed an update (rows affected = 1).
Why would both transactions see `used = 0`? The DB server tries to isolate transactions and actively hides effects of other transactions that have not committed yet.
This is not true in postgres. When the second transaction tries to update the row, it will wait for the first transaction to commit first and then recheck the WHERE.
article says he also uses "order by" clause, but I am wondering if it will severely limit throughput since all messages will need to be sorted on each lookup, but this probably can be solved by introducing index.
It seems strictly worse to use ORDER BY in this case, since if you're using SKIP LOCKED you should be doing parallel processing anyway, and if you're doing parallel processing, ordering is already going out the window.
You have no ordering guarantees, so how can order be important? If 4 work items are scheduled on 4 independent workers, you have no guarantee which will start first or finish first.
I think the order matter at least because you want to have some FIFO approximation, otherwise some tasks can forever stuck in queue and never be picked up.
Unless you can guarantee that the processing time of each job is exactly the same, if you have multiple workers processing the same queue, you can’t order anything except the start time.
You can use locks to effectively break the queue into sub queues so that each sub queue is only being processed by 1 worker. Then you can order that sub queue.
job should be attempted inthe same order/priority they are enqueued, that's the meaning of the word "queue". That they take varrying amounts of time is another matter.
Queue can clearly mean "work that needs to be completed" not necessarily 'work completed in order'. Your definition is much stricter than it needs to be for most use cases.
There is clearly a conceptual difference between a set of things from which you pull things out randomly, and a queue. A queue always has intrinsic criteria to select the next item to be pulled out.
There are many times when the start order doesn’t really matter, and the additional sorting overhead isn’t worth it. In those cases people will still tend to refer to the entity holding the jobs to be processed as a queue despite the fact that it doesn’t strictly follow FIFO order.
If they are being technically precise, queue isn’t the correct term, but language changes with context and time. Either way the implementation isn’t wrong if strict start order has been considered and isn’t important.
I’m not confusing anything. I’ve seen random selection “job queues” implemented many times. As long as you truly don’t care about start order, it’s fine to trade it for increased throughout.
It means that you are telling pg that you don’t care about order, so it is free to optimize the query in whatever way it wants to. The order can change query to query depending on numerous external factors.
I’m not using pg itself as an example. I’m using a specific implementation of a “job queue” built with pg.
I’ve seen and you can search for and find many implementations of “job queues” using relational databases where job start order guarantees are traded away for throughput.
>Applied to job records, this feature enables simple queue processing queries, e.g. SELECT * FROM jobs ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 1.
I recently published a manifesto and code snippets for exactly this in Postgres!
delete from task
where task_id in
( select task_id
from task
order by random() -- use tablesample for better performance
for update
skip locked
limit 1
)
returning task_id, task_type, params::jsonb as params
Holding a transaction open for the duration of a request to an external service makes me nervous. I've seen similar code lock up the database and bring down production. Are you using timeouts and circuit breakers to control the length of the transactions?
Yes, you absolutely need to set a reasonable idle transaction timeout to avoid a disaster (bugs in the code happen) - this can also be done globally in the database settings.
This article was written in 2015, a year before idle_in_transaction_session_timeout parameter was added (in Postgres 9.6) - which is unfortunately still disabled by default, but that's the easiest way to make sure no transaction sits idle for too long.
In my experience, a queue system is the worst thing to find out isn't scaling properly because once you find out your queue system can't architecturally scale, there's no easy fix to avoid data loss. You talk about "several thousand background jobs" but generally, queues are measured in terms of Little's Law [1] for which you need to be talking about rates; according to Little's Law namely average task enqueue rate per second and average task duration per second. Raw numbers don't mean that much.
In the beginning you can do a naive UPDATE ... SET, which locks way too much. While you can make your locking more efficient, doing UPDATE with SELECT subqueries for dequeues and SELECT FOR UPDATE SKIP LOCKED, eventually your dequeue queries will throttle each other's locks and your queue will grind to a halt. You can try to disable enqueues at that point to give your DB more breathing room but you'll have data loss on lost enqueues and it'll mostly be your dequeues locking each other out.
You can try very quickly to shard out your task tables to avoid locking and that may work but it's brittle to roll out across multiple workers and can result in data loss. You can of course drop a random subset of tasks but this will cause data loss. Any of these options is not only highly stressful in a production scenario but also very hard to recover from without a ground-up rearchitecture.
Is this kind of a nightmare production scenario really worth choosing Boring Technology? Maybe if you have a handful of customers and are confident you'll be working at tens of tasks per second forever. Having been in the hot seat for one of these I will always choose a real queue technology over a database when possible.
> and are confident you'll be working at tens of tasks per second forever.
It's more like a few thousand per second, and enqueues win, not dequeues like you say... on very small hardware without tuning. If you're at tens of tasks per second, you have a whole lot of breathing room: don't build for 100x current requirements.
This link is simply raw enqueue/dequeue performance. Factor in workers that perform work or execute remote calls and the numbers change. Also, I find when your jobs have high variance in times, performance degrades significantly.
> This doesn't really make sense to me. To me, the main problem seems to be that you end up with having a lot of snapshots around.
The dequeuer needs to know which tasks to "claim", so this requires some form of locking. Eventually this becomes a bottleneck.
> don't build for 100x current requirements
What happens if you get 100x traffic? Popularity spikes can do it, so can attacks. Is the answer to just accept data loss in those situations? Queue systems are super simple to use. I'm counting "NOTIFY/LISTEN" on Postgres as a queue, because it is a queue from the bottom up.
> Factor in workers that perform work or execute remote calls and the numbers change.
These don't occur on the database server, though... This merely affects the number of rows currently claimed.
> The dequeuer needs to know which tasks to "claim", so this requires some form of locking. Eventually this becomes a bottleneck.
These are just try locks, though-- the row locks are not contended. The big thing you run into is having lots of snapshots around and having to skip a lot of claimed rows for each dequeue.
> What happens if you get 100x traffic? Popularity spikes can do it, so can attacks.
If you get 100x the queueing activity for batch jobs, you're going to have stuff break well before the queue. It's probably not too easy to get 100x the drain rate, even if your queue system can handle it.
This scales well beyond 100M batch tasks per day, which gets you to 1M users with 100 tasks/day each.
NOTIFY/LISTEN isn't a queue it has broadcast semantics. Postgres queueing is really just the SELECT FOR UPDATE SKIP LOCKED, the NOTIFY/LISTEN allows you to reduce the latency a bit but not essential.
Throttle the inputs. Rate-limiting doesn’t belong to the data layer.
While throttling due to organic popularity isn’t great, I’d argue the tradeoffs might be worthwhile. If it looks like the spike will last, stand up Redis during the throttling, double-write, and throttle down the Postgres queue until it’s empty. If you really need to, take a 15 minute outage to just copy data over.
What happens when you get 500x the traffic or 50x?
How does the system behave when the traffic rate is higher for which it was designed for or can currently handle? Because that number will always be there, even in a "scalable" system. One won't be able to add capacity at the same rate that work will increase.
You are going to have the same scaling issues with your datastore. I don't really understand why you say that your dequeue queries will throttle each others locks and grind it to a half? Isn't that the whole point of SKIP LOCKED?
If you find yourself in that situation, migrating to a more performant queuing solution is not that much of a leap. You already have an overall system architecture that scales well (async processing with a queue).
_Ideally_ the queuing technology is abstracted from the job-submitters/job-runners anyway. It's a bit more work if multiple services are just writing to the queue table directly.
I agree that the _moment_ the system comes to a screeching halt is definitely not fun.
Skip locked is useful till you have to maintain order for a group of messages with some "group_id", so that set of related messages are sent one after the other.
Then you probably have to write complicated queries or use partitions in some sort.
Also, postgres partial indexes can be quite helpful in situations where you want to persist and query intermediate job lifecycle state and don't want multiple rows or tables to track one type of job queue
One issue with Redis as a queue backend seems to be that persistence is quite expensive, at least for managed Redis instances. Using PG seems like it could be much cheaper, especially if you already have an instance with room to spare.
I thought it was an interesting article, and I'd love to hear more from people using PG for queues in production (my intuition would say you'd get a lot of table bloat and/or vacuum latency, but I haven't tested it myself), but when it comes to the conclusion - "choosing boring technology should be one’s default choice" - I can't think of anything more boring (in a good sense, mostly) than Sidekiq + Redis for a Rails app.
Same here. Sidekiq + Rails in a Rails app is a powerhouse, simple and reliable, but I do worry about losing the queue in Redis. It would be great to have that in Postgres as well.
For running queues on Postgres with Node.js backend(s), I highly recommend https://github.com/timgit/pg-boss. I'm sure it has it scale limits. But if you're one of the 90% of the apps that never needs any kind of scale that a modern server can't easily handle then it's fantastic. You get transactional queueing of jobs, and it automatically handles syncing across multiple job processing servers using Postgres locks.
Another point is that task queue technology is highly fungible. There's nothing stopping you from starting with cron, adding in Postgres/Redis, then graduating to Kafka or something as need arises. And running all three in parallel, with different jobs in each. I would be surprised if the average Kafka shop didn't also have a bunch of random cron jobs doing things that could be implemented on Kafka or vice versa.
At some point you may want to refactor things to reduce tech debt, but it really is a "and" rather than "or" decision.
I agree with all of this except for the part about “cron”. Cron jobs in my experience quickly become hard to manage and effectively invisible over time.
Use almost anything else to manage job scheduling….
Running this exact implementation with 47M jobs processed and counting. SKIP LOCKED is great for VACUUM, and having durable storage with indexes make otherwise expensive patterns like delayed jobs, retries, status updates, "at least once", etc. really easy to implement.
I'm sure Redis is much faster than an RDBMS w/ all the ACID features turned on. The biggest concern I always have with Redis is simply overwhelming the in-memory storage limits when someone wants to do process a large number of good-sized messages at an inconvenient time. #tradeoffs
I do enjoy using https://github.com/graphile/worker for my postgresql queuing needs. Very scalable, the next release 0.14 even more so, and easy to use.
Maybe it's changed in the last year or so, but from benchmarking and writing / running queue software for Postgres - SKIP LOCKED was/is significantly faster. Is that different for MySQL?
Pretty common advice for scaling Postgres is to deploy pgbouncer in transaction mode in front of it to handle connection pooling.
Advisory locks don’t work in this setup (and will start behaving in strange ways if you do try to use them.) Something to consider if you go this route.
To do anything safe and interesting you’ll need transactions. Using SKIP LOCKED won’t be your bottleneck, your application will. Job queues are about side effects and the rest of your application needs to keep up.
One of the biggest benefits imo of using Postgres as your application queue, is that any async work you schedule benefits from transactionality.
That is, say you have a relatively complex backend mutation that needs to schedule some async work (eg sending an email after signup). With a Postgres queue, if you insert the job to send the email and then in a later part of the transaction, something fails and the transaction rollbacks, the email is never queued to be sent.
Good point. We alleviate that a bit by scheduling our queue adds to not run until after commit. But then we still have some unsafety, and if connection to rabbit is down we're in trouble.
Worth being clear that bridging to another non-idempotent system necessarily requires you to pick at-least-once or at-most-once semantics. So for emails, if you fail awaiting confirmation of your email you still need to pick between failing your transaction and potentially duplicating the email, or continuing and potentially dropping it.
The big advantage is for code paths which async modify your DB; these can be done fully transactionally with exactly-once semantics since the Job consumption and DB update are in the same transaction.
That's kind of missing the parent's point. If you wanted to ensure emails arrive, that sounds like another queue that could be backed by a different table that is also produced into as part of the original transaction.
> One of the biggest benefits imo of using Postgres as your application queue, is that any async work you schedule benefits from transactionality.
This is a really important point. I often end up using a combination of Postgres and SQS since SQS makes it easy to autoscale the job processing cluster.
In Postgres I have a transaction log table that includes columns for triggered events and the pg_current_xact_id() for the transaction. (You can also use the built in xmin of the row but then you have to worry about transaction wrap around.) Inserting into this row triggers a NOTIFY.
A background process runs in a loop. Selects all rows in the transaction table with a transaction id between the last run's xmin and the current pg_snapshot_xmin(pg_current_snapshot()). Maps those events to jobs and submits them to SQS. Records the xmin. LISTEN's to await the next NOTIFY.
I'm not sure this is really an issue with transactionality as a single request can obviously be split up into multiple transactions, but rather that even if you correctly flag the email as pending/errored, you either need to process these manually, or have some other kind of background task that looks for them, at which point why not just process them asynchronously.
Another benefit of this is that you're guaranteed that the transaction is completed before the job is picked up. With redis-backed queues (or really anything else), you very quickly run into the situation where your queue executes a job depending on a database record existing prior to the transaction being committed (and the fix for this is usually awkward / complex code).
> With a Postgres queue, if you insert the job to send the email and then in a later part of the transaction, something fails and the transaction rollbacks, the email is never queued to be sent.
An option could be use a second connection and a separate transaction to insert data in the queue table.
I agree - having to tell a database that something was processed, and fire off a message into RabbitMQ, say, is never 100% transactional. This would be my top reason to use this approach.
> With a Postgres queue, if you insert the job to send the email and then in a later part of the transaction, something fails and the transaction rollbacks, the email is never queued to be sent.
This is true - definitely worth isolating what should be totally separate database code into different transactions. On the other hand, if your user is not created in the DB, you might not want your signup email. Just depends on the situation.
It seems like listen/notify doesn't play well with a serverless architecture. Would it make sense for Postgres to make a web request when there's work in the queue? Is that a thing?
1. The main downside to using PostgreSQL as a pub/sub bus with LISTEN/NOTIFY is that LISTEN is a session feature, making it incompatible with statement level connection pooling.
2. If you are going to do this use advisory locks [0].
Other forms of explicit locking put more pressure on the database while advisory locks are deliberately very lightweight.
My favorite example implementation is que [1] which is ported to several languages.
At my previous company, they switched from using NOTIFY/LISTEN for Postgres notifications to a custom solution built on top of logical replication. As I understand it, part of the reason was reliability. I never touched that part of the code, but I believe the idea was to subscribe to logical replication updates and send out notifications based on these.
Alternatively if you just want to quickly hack something into your application, here is a complete solution in one Python function with retries (ask ChatGPT to tell you what the table structure is):
import psycopg2
import psycopg2.extras
import random
db_params = {
'database': 'jobs',
'user': 'jobsuser',
'password': 'superSecret',
'host': '127.0.0.1',
'port': '5432',
}
conn = psycopg2.connect(**db_params)
cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
def do_some_work(job_data):
if random.choice([True, False]):
print('do_some_work FAILED')
raise Exception
else:
print('do_some_work SUCCESS')
def process_job():
sql = """DELETE FROM message_queue
WHERE id = (
SELECT id
FROM message_queue
WHERE status = 'new'
ORDER BY created ASC
FOR UPDATE SKIP LOCKED
LIMIT 1
)
RETURNING *;
"""
cur.execute(sql)
queue_item = cur.fetchone()
print('message_queue says to process job id: ', queue_item['target_id'])
sql = """SELECT * FROM jobs WHERE id =%s AND status='new_waiting' AND attempts <= 3 FOR UPDATE;"""
cur.execute(sql, (queue_item['target_id'],))
job_data = cur.fetchone()
if job_data:
try:
do_some_work(job_data)
sql = """UPDATE jobs SET status = 'complete' WHERE id =%s;"""
cur.execute(sql, (queue_item['target_id'],))
except Exception as e:
sql = """UPDATE jobs SET status = 'failed', attempts = attempts + 1 WHERE id =%s;"""
# if we want the job to run again, insert a new item to the message queue with this job id
cur.execute(sql, (queue_item['target_id'],))
else:
print('no job found, did not get job id: ', queue_item['target_id'])
conn.commit()
process_job()
cur.close()
conn.close()
Yep, we process hundreds of thousands and sometimes a few million jobs daily inside Postgres, using Oban in Elixir.
Having transactional semantics around background jobs is incredibly convenient for things like scheduling email only if the transaction is successful, and so on.
You do need to do a little bit of autovacuum tuning, but once sorted it’s been great for us.
there's an important dimension of scalability that I think gets overlooked in a lot of these discussions about database-as-a-queue vs queue-system-as-a-queue:
are you queuing jobs, or are you queuing messages?
that's a fuzzy distinction, so somewhat equivalently, what's the expected time it takes for a worker to process a given queue item?
at one end, an item on the queue may take several seconds to a minute or longer to process. at the other end, an item might take only a few milliseconds to process. in that latter case, it's often useful to do micro-batching, where a single worker pulls 100 or 1000 items off the queue at once, and processes them as a batch (such as by writing them to a separate datastore)
the "larger" the items are (in terms of wall-clock processing time, not necessarily in terms of size in bytes of the serialized item payload) the more effective the database-as-a-queue solution is, in my experience.
as queue items get smaller / shorter to process, and start to feel more like "messages" rather than discrete "jobs", that's when I tend to reach for a queue system over a database.
for example, there's a RabbitMQ blog post [0] on cluster sizing where their recommendations start at 1000 messages/second. that same message volume on a database-as-a-queue would require, generally speaking, 3000 write transactions per second (if we assume one transaction to enqueue the message, one for a worker to claim it, and one for a worker to mark it as complete / delete it).
can Postgres and other relational databases be scaled & tuned to handle that write volume? yes, absolutely. however, how much write volume are you expecting from your queue workload, compared to the write volume from its "normal database" workload? [1]
I think that ends up being a useful heuristic when deciding whether or not to use a database-as-a-queue - will you have a relational database with a "side gig" of acting like a queue, or will you have a relational database that in terms of data volume is primarily acting like a queue, with "normal database" work relegated to "side gig" status?
1: there's also a Postgres-specific consideration here where a lot of very short-lived "queue item" database rows can put excessive pressure on the autovacuum system.
I’ve used PG as a message queue, actually it was used as a transactional front end to Kafka; we’d push messages to a PG table during a transaction, which would then be snarfed up to Kafka by a separate process after the transaction completed.
I’ve seen very high transaction rates from this arrangement, more than 20k messages/second.
I maintain QueueClassic (https://github.com/QueueClassic/queue_classic) for Rails/Ruby folks; which is basically what you're talking about - a queuing system for Postgres. A bonus reason, and why I originally wanted this was the ability to use transactions fully - i.e. I can start one, do some stuff, add a job in to the queue (to send an email), .....and either commit, or roll back - avoiding sending the email. If you use resque, I found sometimes either you can't see the record (still doing other stuff and it's not committed), or it's not there (rollback) - so either way you had to deal with it.
QC (and equivs) use the same db, and same connection, so same transaction. Saves quite a bit of cruft.
You don't even need a database to make a message queue. The Linux file system makes a perfectly good basis for a message queue since file moves are atomic.
My guess is that many people are implementing queuing mechanisms just for sending email.
You can see how this works in Arnie SMTP buffer server, a super simple queue just for emails, no database at all, just the file system.
This is true, and I’ve worked on systems that use this, but it’s a lot more work than just a rename.
I’d recommend that, if you have a Postgres database already, definitely use that instead. Your queues will be transactional and they will get backed up when the rest of your database does.
Well, if you have multiple writers then you need to decide who’s responsible for rotating the queues, and you need to serialise writes; or if each writer has its own queue then the reader has to do more work. And then you need to worry about fsync, and backup. And of course you need to be careful to flush to the queue after each write to avoid partial writes.
Basically I’m saying that there are just a number of potential footguns when using files as queues - I speak from experience! - which are trivially taken care of by a database, especially if you have one already.
I’m not saying that it’s not possible, just that for non trivial applications, it’s certainly more complex than just an atomic file move.
For those using simple SELECTs, what kind of WHERE clause are you using that works well with lots of qualified pending messages and (somewhat) guarantees the most appropriate (oldest?) message?
We use exactly this for windmill (OSS Retool alternative + modern airflow) and run benchmarks everyday. On a modest github CI instance where one windmill worker and postgres run as containers, our benchmarks run at 1200jobs/s. Workers can be added and it will scale gracefully up to 5000jobs/s. We are exploring using Citus to cross the barrier of 5000j/s on our multi-tenant instance.
I'm always surprised that when I see people talk about queues I never see anyone mention beanstalkd. I've been using it for basically everything for 10 years and it's solid as a rock, incredibly simple and requires basically no maintenance. It Just Works™
Yes, I liked it when I encountered it a few years back. Lately been using redis and rq on a recent project since we're already using redis for caching. Is there much difference other than that?
I've implemented queues with tables in RDBMSs a few times and it's always great and usually all you need. Worried about future scale? Make a class to wrapper the queue with a decent interface and swap it for RabbitMQ or whatever you want down the road. Implementation stays opaque and you have an easy upgrade path later on.
Temporal, which AFAIK was made by the Uber Cadence team, which was also involved in SQS, uses postgres as a backend.
I used it for a web automation system for an accounting client (automatically read files from a network share, lookup the clients on a database, submit the documents to government websites, using headless browsers, and put the resulting files in the directory). It allows for completely effortless deterministic programs that call workers that run the non deterministic code, with built in configurable retries (react to certain exception type, exponential back off) so you can write code that works almost like there were no issues with api connections, filesystem, etc.
This code has been running for 5 or more years, with barely any maintenance, with 0 issues so far. It keeps everything in postgres, so even full reboots and crashes have no impact, it will just move the work back to the queue and it will run when there's an available worker.
Temporal is a pretty complicated system. It has sharding built in, stores the entire activity history and runs multiple queues for timers and events. I’m a big fan (worked at Uber) but it’s definitely not just postgres with a few indices.
It's about the challenge of matching up transactions with queues - where you want a queue to be populated reliably if a transaction completes, and also reliably NOT be populated if it doesn't.
Brandur's pattern is to have an outgoing queue in a database table that gets updated as part of that transaction, and can then be separately drained to whatever queue system you like.
We used postgres for some of our queues back when we were at ~10 msg/s. It scaled quite a bit, but, honestly, setting up SQS or some other queue stack in AWS, GCP, or Azure is so simple and purpose built for the task (with DL queues and the like built in), I don’t know why you wouldn’t just go that route and not have to worry about that system shitting the bed and affecting the rest of the DB’s health.
It seems foolish. I am a big fan of “use the dumbest tool”, but sometimes engineers take it too far and you’re left with the dumbest tool with caveats that don’t seem worth it given the mainstream alternative is relatively cheap and simple.
You can't do better than At Least Once if you're having side effects outside the database, so it's not clear that SQS's weaker semantics have any practical effect.
What I've settled on is "store most job state in the DB, use task queues just to poke workers into working on the jobs".
Storing the job state in the DB means you can query state nicely. It's not going to exactly show the state of things but it's helpful for working through a live incident (especially when most job queues just delete records as work is processed).
And if you make all the background tasks idempotent anyways then you're almost always safe with running a thing like "send a job to the task queue to handle this job".
If you rely _just_ on message queues, there are a lot of times where you can have performance issues, yet have a lot of trouble knowing what's going on (for example, rabbitMQ might tell you the size of your queues, but offer little-to-no inspection of the data inside them).
Ultimately you have to figure out the separation of concerns of the job state and other core state. Ranging from “all state stored in message and will never become out of sync” to “no state stored in message and will never become out of sync”. In between you have “some state stored in db and some in message” and what I’ve found to be useful is keeping stuff in the db that needs to have high end state integrity (or as you said just making sure jobs are cancellable/idempotent).
Tangible example:
We have a video transcoder queue. The state of the video model in our db can change as the video is being finalized in various ways. The transcoder generates thumbnails and assets from the video and also updates its state in the db. So we store job information in the message about what thumbnails we want to generate and the video ID but nothing else. This allows us to look up the video row, see if the same media was already transcoded from the video (and cancel the job), and, if not, run the job and update the video row.
Also (and I know you’re not saying this), but I’ve never understood the argument that keeping queues in Postgres leads to higher data integrity via transaction guarantees. The job is still running on another process outside of the db. The only time this could be true is if the job itself mostly updates state in the db, in which case it’s the small minority of queued workloads (with the majority needing to do non-db compute work).
> I don’t know why you wouldn’t just go that route and not have to worry about that system shitting the bed
Because different software has different requirements. Not having an external service requirement other than Postgres might be a feature of an on-prem/b2b appliance.
Because some software may be projected never outgrow the capabilities of Postgres, and if it does moving to another service can be made very easy.
Because you want a transitional job system and the simplicity of doing it in Postgres.
368 comments
[ 4.8 ms ] story [ 569 ms ] threadhttps://www.pgcasts.com/episodes/the-skip-locked-feature-in-...
It’s not “web scale” but it easily extends to several thousand background jobs in my experience
I've used this for tasks at big organizations without issue. No need for any special deployments or new infra. Just spin up a few worker threads in your app. Perhaps a thread to reset abandoned tasks. But in three years this never actually happened, as everything was contained in try/catch that would add it back to the queue, and our java app was damn stable.
Just curious. We maintained a custom background processing system for years but recently replaced it with off the shelf stuff, so I'm really interested in how others are doing similar stuff.
Our tasks were quick enough so that all fetched tasks would always be able to be completed before a scale down / new deploy etc, but we stopped fetching new ones when the signal came so it just finished what it had. I updated above, we did have logic to monitor if a task got taken but never got a finished status, but I can't remember it ever actually reporting on anything.
When grabbing a new message it selects "Available or (Locked with Updated timestamp older than configured timeout)". If successful it immediately tries to set the Locked status, Updated timestamp and bumps the Version counter, where the previous values of Status and Version has to match. If the update fails it retries getting a new message.
If the Version counter is too high, it moves the message to the associated dead-letter table, and retries getting a new message.
This isn't for high performance. I tested it and got 1000 messages/sec throughput with handful of producers and consumers against test db instance (limited hardware), which would be plenty for us.
I wrote it to be simple and so we could easily move to something AMPQ'ish like RabbitMQ or Azure Service Bus when needed. Overall quite easy to implement and has served us well so far.
My approach was:
- Accept the inbound call
- Generate a 20 character random string (used as a signature)
- Execute a sql query that selects the oldest job without a signature and write the signature, return the primary key of the job that was updated.
- If it errors for any reason, loop back and attempt again, but only 10 times, as some underlying issue exists (10 collisions is statistically improbable for my use case)
- Read the primary key returned by that sql query and read it, comparing it's signature to my random one.
- If a hit, return the job to the caller
- If a miss, loop back and start again, incrementing attempts by 1.
The caller has to handle the possibility that a call to this web service won't return anything, either due to no jobs existing, or the collision/error threshold being reached.
In either case, the caller backs for it's configured time, then calls again.
Callers are usually in 'while true' loops, only existing if they get an external signal to close or an uncontrolled crash.
If you take this approach, you will have a function or a web service that converts the SQL table into a job queue service. When you do that, you can build metrics on the amount of collisions you get while trying to pull and assign jobs to workers.
I had inbuilt processes that would sweep through jobs that were assigned (had a job signature) and weren't marked as complete, it actioned those to handle the condition of a crashed worker.
There are many many other services the proper job queues offer, but that usually means more dependencies, and code libraries / containers, so just build in the functionality you need.
If it is accurate, fast enough, and stable, you've got the best solution for you.
/edited for formatting
https://stackoverflow.com/a/76821755
There is actually another possibility: there must be a way to check whether the receiving system has received the message. But this only works if there are no "rogue" senders.
Edit: To clarify, I mean `SELECT id WHERE used = 0` followed by `UPDATE ... SET used = 1 WHERE id = ... AND used = 0`
Why would both transactions see `used = 0`? The DB server tries to isolate transactions and actively hides effects of other transactions that have not committed yet.
https://www.postgresql.org/docs/current/transaction-iso.html...
You can use locks to effectively break the queue into sub queues so that each sub queue is only being processed by 1 worker. Then you can order that sub queue.
If they are being technically precise, queue isn’t the correct term, but language changes with context and time. Either way the implementation isn’t wrong if strict start order has been considered and isn’t important.
I care about money, I dont have money.
I’m not confusing anything. I’ve seen random selection “job queues” implemented many times. As long as you truly don’t care about start order, it’s fine to trade it for increased throughout.
Does that mean it doesn't have any order or that whoever writes the query doesn't care about order?
Also we are arguing over whether pg suffices as a queue implementation, and you use itself as an example?
I’m not using pg itself as an example. I’m using a specific implementation of a “job queue” built with pg.
I’ve seen and you can search for and find many implementations of “job queues” using relational databases where job start order guarantees are traded away for throughput.
That's exactly what a queue means, not just in every day life, but specifically in computer science.
Queue: a list of data items, commands, etc., stored so as to be retrievable in a definite order, usually the order of insertion.
note the term "Usually", not "always".
always has an oder, which is usually of insertion.
>Applied to job records, this feature enables simple queue processing queries, e.g. SELECT * FROM jobs ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 1.
> The task row will not be deleted if sendEmail fails. The PG transaction will be rolled back. The row and sendEmail will be retried.
For example, I run the above query to grab a queued email, send it using mailgun, then COMMIT. Nothing is changed in the DB unless the email is sent.
Careful monitoring and tuning of parameters mentioned by the sibling comment to you can help mitigate this, though.
Ultimately at scale, no, RDBMS shouldn’t be a queue. But most have a long way to go before they hit that point.
In the beginning you can do a naive UPDATE ... SET, which locks way too much. While you can make your locking more efficient, doing UPDATE with SELECT subqueries for dequeues and SELECT FOR UPDATE SKIP LOCKED, eventually your dequeue queries will throttle each other's locks and your queue will grind to a halt. You can try to disable enqueues at that point to give your DB more breathing room but you'll have data loss on lost enqueues and it'll mostly be your dequeues locking each other out.
You can try very quickly to shard out your task tables to avoid locking and that may work but it's brittle to roll out across multiple workers and can result in data loss. You can of course drop a random subset of tasks but this will cause data loss. Any of these options is not only highly stressful in a production scenario but also very hard to recover from without a ground-up rearchitecture.
Is this kind of a nightmare production scenario really worth choosing Boring Technology? Maybe if you have a handful of customers and are confident you'll be working at tens of tasks per second forever. Having been in the hot seat for one of these I will always choose a real queue technology over a database when possible.
[1]: https://en.wikipedia.org/wiki/Little%27s_law
It's more like a few thousand per second, and enqueues win, not dequeues like you say... on very small hardware without tuning. If you're at tens of tasks per second, you have a whole lot of breathing room: don't build for 100x current requirements.
https://chbussler.medium.com/implementing-queues-in-postgres...
> eventually your dequeue queries will throttle each other's locks a
This doesn't really make sense to me. To me, the main problem seems to be that you end up with having a lot of snapshots around.
This link is simply raw enqueue/dequeue performance. Factor in workers that perform work or execute remote calls and the numbers change. Also, I find when your jobs have high variance in times, performance degrades significantly.
> This doesn't really make sense to me. To me, the main problem seems to be that you end up with having a lot of snapshots around.
The dequeuer needs to know which tasks to "claim", so this requires some form of locking. Eventually this becomes a bottleneck.
> don't build for 100x current requirements
What happens if you get 100x traffic? Popularity spikes can do it, so can attacks. Is the answer to just accept data loss in those situations? Queue systems are super simple to use. I'm counting "NOTIFY/LISTEN" on Postgres as a queue, because it is a queue from the bottom up.
These don't occur on the database server, though... This merely affects the number of rows currently claimed.
> The dequeuer needs to know which tasks to "claim", so this requires some form of locking. Eventually this becomes a bottleneck.
These are just try locks, though-- the row locks are not contended. The big thing you run into is having lots of snapshots around and having to skip a lot of claimed rows for each dequeue.
> What happens if you get 100x traffic? Popularity spikes can do it, so can attacks.
If you get 100x the queueing activity for batch jobs, you're going to have stuff break well before the queue. It's probably not too easy to get 100x the drain rate, even if your queue system can handle it.
This scales well beyond 100M batch tasks per day, which gets you to 1M users with 100 tasks/day each.
Throttle the inputs. Rate-limiting doesn’t belong to the data layer.
While throttling due to organic popularity isn’t great, I’d argue the tradeoffs might be worthwhile. If it looks like the spike will last, stand up Redis during the throttling, double-write, and throttle down the Postgres queue until it’s empty. If you really need to, take a 15 minute outage to just copy data over.
This line of reasoning is desirable for FAANGS, but can bankrupt startups that need to move fast and get shit done.
How does the system behave when the traffic rate is higher for which it was designed for or can currently handle? Because that number will always be there, even in a "scalable" system. One won't be able to add capacity at the same rate that work will increase.
_Ideally_ the queuing technology is abstracted from the job-submitters/job-runners anyway. It's a bit more work if multiple services are just writing to the queue table directly.
I agree that the _moment_ the system comes to a screeching halt is definitely not fun.
Then you probably have to write complicated queries or use partitions in some sort.
Or Just stick to one thread polling the messages.
Also, postgres partial indexes can be quite helpful in situations where you want to persist and query intermediate job lifecycle state and don't want multiple rows or tables to track one type of job queue
I thought it was an interesting article, and I'd love to hear more from people using PG for queues in production (my intuition would say you'd get a lot of table bloat and/or vacuum latency, but I haven't tested it myself), but when it comes to the conclusion - "choosing boring technology should be one’s default choice" - I can't think of anything more boring (in a good sense, mostly) than Sidekiq + Redis for a Rails app.
Python has Celery, but maybe the author is looking for more choice between brokers. https://docs.celeryq.dev/en/stable/index.html
At some point you may want to refactor things to reduce tech debt, but it really is a "and" rather than "or" decision.
Use almost anything else to manage job scheduling….
Do you also have any idea on the concurrency? How many workers you have pulling from Postgres.
I’ve used this approach before (ages ago) when Redis wasn’t even a thing, though not at high throughout requirements.
https://github.com/bensheldon/good_job
Had it in production for about a quarter and it’s worked well.
Do not use SKIP LOCKED unless it is a toy/low throughout.
Row locks require transactions and disk writes.
Advisory locks require neither. (However, you do have to stay inside the configurable memory budget.)
Advisory locks don’t work in this setup (and will start behaving in strange ways if you do try to use them.) Something to consider if you go this route.
Oban is able to run over 1m jobs a minute, and the ultimate bottleneck is throttling in application code to prevent thrashing the database: https://getoban.pro/articles/one-million-jobs-a-minute-with-...
However in the empty poll case, you can avoid a transaction.
IDK maybe <1000 messages per minute
Not saying SKIP LOCKED can't work with that many. But you'll probably want to do something with lower overhead.
FWIW, Que uses advisory locks [1]
[1] https://github.com/que-rb/que
That is, say you have a relatively complex backend mutation that needs to schedule some async work (eg sending an email after signup). With a Postgres queue, if you insert the job to send the email and then in a later part of the transaction, something fails and the transaction rollbacks, the email is never queued to be sent.
The big advantage is for code paths which async modify your DB; these can be done fully transactionally with exactly-once semantics since the Job consumption and DB update are in the same transaction.
This is a really important point. I often end up using a combination of Postgres and SQS since SQS makes it easy to autoscale the job processing cluster.
In Postgres I have a transaction log table that includes columns for triggered events and the pg_current_xact_id() for the transaction. (You can also use the built in xmin of the row but then you have to worry about transaction wrap around.) Inserting into this row triggers a NOTIFY.
A background process runs in a loop. Selects all rows in the transaction table with a transaction id between the last run's xmin and the current pg_snapshot_xmin(pg_current_snapshot()). Maps those events to jobs and submits them to SQS. Records the xmin. LISTEN's to await the next NOTIFY.
An option could be use a second connection and a separate transaction to insert data in the queue table.
> With a Postgres queue, if you insert the job to send the email and then in a later part of the transaction, something fails and the transaction rollbacks, the email is never queued to be sent.
This is true - definitely worth isolating what should be totally separate database code into different transactions. On the other hand, if your user is not created in the DB, you might not want your signup email. Just depends on the situation.
1. The main downside to using PostgreSQL as a pub/sub bus with LISTEN/NOTIFY is that LISTEN is a session feature, making it incompatible with statement level connection pooling.
2. If you are going to do this use advisory locks [0]. Other forms of explicit locking put more pressure on the database while advisory locks are deliberately very lightweight.
My favorite example implementation is que [1] which is ported to several languages.
[0] https://www.postgresql.org/docs/current/explicit-locking.htm...
[1] https://github.com/que-rb/que
Recently I had to stop using it because after a while all NOTIFY/LISTENS would stop working, and only a database restart would fix the issue https://dba.stackexchange.com/questions/325104/error-could-n...
https://github.com/cpursley/walex
I built a complete implementation in Python designed to work the same as SQS but be more simple:
https://github.com/starqueue/starqueue
Alternatively if you just want to quickly hack something into your application, here is a complete solution in one Python function with retries (ask ChatGPT to tell you what the table structure is):
Having transactional semantics around background jobs is incredibly convenient for things like scheduling email only if the transaction is successful, and so on.
You do need to do a little bit of autovacuum tuning, but once sorted it’s been great for us.
are you queuing jobs, or are you queuing messages?
that's a fuzzy distinction, so somewhat equivalently, what's the expected time it takes for a worker to process a given queue item?
at one end, an item on the queue may take several seconds to a minute or longer to process. at the other end, an item might take only a few milliseconds to process. in that latter case, it's often useful to do micro-batching, where a single worker pulls 100 or 1000 items off the queue at once, and processes them as a batch (such as by writing them to a separate datastore)
the "larger" the items are (in terms of wall-clock processing time, not necessarily in terms of size in bytes of the serialized item payload) the more effective the database-as-a-queue solution is, in my experience.
as queue items get smaller / shorter to process, and start to feel more like "messages" rather than discrete "jobs", that's when I tend to reach for a queue system over a database.
for example, there's a RabbitMQ blog post [0] on cluster sizing where their recommendations start at 1000 messages/second. that same message volume on a database-as-a-queue would require, generally speaking, 3000 write transactions per second (if we assume one transaction to enqueue the message, one for a worker to claim it, and one for a worker to mark it as complete / delete it).
can Postgres and other relational databases be scaled & tuned to handle that write volume? yes, absolutely. however, how much write volume are you expecting from your queue workload, compared to the write volume from its "normal database" workload? [1]
I think that ends up being a useful heuristic when deciding whether or not to use a database-as-a-queue - will you have a relational database with a "side gig" of acting like a queue, or will you have a relational database that in terms of data volume is primarily acting like a queue, with "normal database" work relegated to "side gig" status?
0: https://blog.rabbitmq.com/posts/2020/06/cluster-sizing-and-o...
1: there's also a Postgres-specific consideration here where a lot of very short-lived "queue item" database rows can put excessive pressure on the autovacuum system.
I’ve seen very high transaction rates from this arrangement, more than 20k messages/second.
QC (and equivs) use the same db, and same connection, so same transaction. Saves quite a bit of cruft.
My guess is that many people are implementing queuing mechanisms just for sending email.
You can see how this works in Arnie SMTP buffer server, a super simple queue just for emails, no database at all, just the file system.
https://github.com/bootrino/arniesmtpbufferserver
I’d recommend that, if you have a Postgres database already, definitely use that instead. Your queues will be transactional and they will get backed up when the rest of your database does.
Such as?
Basically I’m saying that there are just a number of potential footguns when using files as queues - I speak from experience! - which are trivially taken care of by a database, especially if you have one already.
I’m not saying that it’s not possible, just that for non trivial applications, it’s certainly more complex than just an atomic file move.
https://github.com/windmill-labs/windmill/tree/benchmarks
For one such project, the message 'priority' feature was a life saver and a feature that is not super common in competing solutions.
I used it for a web automation system for an accounting client (automatically read files from a network share, lookup the clients on a database, submit the documents to government websites, using headless browsers, and put the resulting files in the directory). It allows for completely effortless deterministic programs that call workers that run the non deterministic code, with built in configurable retries (react to certain exception type, exponential back off) so you can write code that works almost like there were no issues with api connections, filesystem, etc.
This code has been running for 5 or more years, with barely any maintenance, with 0 issues so far. It keeps everything in postgres, so even full reboots and crashes have no impact, it will just move the work back to the queue and it will run when there's an available worker.
Transactionally Staged Job Drains in Postgres - https://brandur.org/job-drain
It's about the challenge of matching up transactions with queues - where you want a queue to be populated reliably if a transaction completes, and also reliably NOT be populated if it doesn't.
Brandur's pattern is to have an outgoing queue in a database table that gets updated as part of that transaction, and can then be separately drained to whatever queue system you like.
It seems foolish. I am a big fan of “use the dumbest tool”, but sometimes engineers take it too far and you’re left with the dumbest tool with caveats that don’t seem worth it given the mainstream alternative is relatively cheap and simple.
Storing the job state in the DB means you can query state nicely. It's not going to exactly show the state of things but it's helpful for working through a live incident (especially when most job queues just delete records as work is processed).
And if you make all the background tasks idempotent anyways then you're almost always safe with running a thing like "send a job to the task queue to handle this job".
If you rely _just_ on message queues, there are a lot of times where you can have performance issues, yet have a lot of trouble knowing what's going on (for example, rabbitMQ might tell you the size of your queues, but offer little-to-no inspection of the data inside them).
Tangible example:
We have a video transcoder queue. The state of the video model in our db can change as the video is being finalized in various ways. The transcoder generates thumbnails and assets from the video and also updates its state in the db. So we store job information in the message about what thumbnails we want to generate and the video ID but nothing else. This allows us to look up the video row, see if the same media was already transcoded from the video (and cancel the job), and, if not, run the job and update the video row.
Also (and I know you’re not saying this), but I’ve never understood the argument that keeping queues in Postgres leads to higher data integrity via transaction guarantees. The job is still running on another process outside of the db. The only time this could be true is if the job itself mostly updates state in the db, in which case it’s the small minority of queued workloads (with the majority needing to do non-db compute work).
Because different software has different requirements. Not having an external service requirement other than Postgres might be a feature of an on-prem/b2b appliance.
Because some software may be projected never outgrow the capabilities of Postgres, and if it does moving to another service can be made very easy.
Because you want a transitional job system and the simplicity of doing it in Postgres.