8 ms·

I use Postgres SKIP LOCKED as a queue. I used to use SQS but Postgres gives me everything I want. I can also do priority queueing and sorting. I gave up on S

I use Postgres SKIP LOCKED as a queue. I used to use SQS but Postgres gives me everything I want. I can also do priority queueing and sorting. I gave up on SQS when it couldn't be accessed from a VPC. AWS might have fixed that now. All the other queueing mechanisms I investigated were dramatically more complex and heavyweight than Postgres SKIP LOCKED.

varelaz 7y ago
There are VPC API endpoints for this. As far as I recall it was all the time there from VPC begging, but I might be wrong.
- andrewstuart 7y ago
  Only since December 2018 and you need to pay extra... https://aws.amazon.com/about-aws/whats-new/2018/12/amazon-sqs-vpc-endpoints-aws-privatelink/ https://aws.amazon.com/about-aws/whats-new/2018/12/amazon-sq...
LunaSea 7y ago
Do you have an example of such a queue somewhere? I think I have a rough idea about how it works because I implemented something similar about four years ago in PostgreSQL but kept getting locking issues I couldn't get out of: - https://stackoverflow.com/questions/33467813/concurrent-crawler-postgresql-query-issues https://stackoverflow.com/questions/33467813/concurrent-craw... - https://stackoverflow.com/questions/29807033/postgresql-concurrent-transaction-issues https://stackoverflow.com/questions/29807033/postgresql-conc... Also, what kind of queue size / concurrency on the polling side are you able to sustain for your current hardware?
- sorokod 7y ago
  Here https://www.2ndquadrant.com/en/blog/what-is-select-skip-locked-for-in-postgresql-9-5/ https://www.2ndquadrant.com/en/blog/what-is-select-skip-lock... is a good discussion.
- andrewstuart 7y ago
  Refer example code above. I don't think Postgres included SKIP LOCKED 4 years ago.
- floatboth 7y ago
  There's a few libraries implementing this: https://github.com/mbuhot/ecto_job https://github.com/mbuhot/ecto_job for Elixir https://github.com/timgit/pg-boss https://github.com/timgit/pg-boss for Node.js
sorokod 7y ago
Using SKIP LOCKED - do you commit the change to the dequeued item (ack it) at the point where you exit the DB call. If so what happens if the instance that dequeued the messages crashes?
- napsterbr 7y ago
  Not GP but I think this wouldn't be a problem. The consumer dequeues with SELECT FOR UPDATE within a transaction. If it crashes the database would rollback the transaction, and then another consumer would be able to select the work unit. As for acking, I see two common methods: using an additional boolean column, something like is_processed. Consumers skip truthy ones. Or, after the work is done, simply delete the entry or move elsewhere (e.g. For archival / auditing).
  - sorokod 7y ago
    My question assumed a scenario where a consumer dequeues a batch, commits the deqieued change, and then crashes while processing the batch. Offcourse one could delay the commit until all processing is completed but then reasoning about the queue throughput becomes tricky.
    - napsterbr 7y ago
      That's the challenge of distributed systems :) it really boils down to how you want failures to be handled. If you ack before processing, and then you crash, those messages are lost (assuming you can't recover from the crash and you are not using something like a two-phase commit). If you ack after processing, you may fail after the messages have been processed but before you've been able to ack them. This leads to duplicates, in which case you better hope your work units are idempotent. If they are not, you can always keep a separate table of message IDs that have been processed, and check against it. Either way, it's hard, complex and there are thousands of intermediate failure cases you have to think about. And for each possible solution (2pc, separate table of message IDs for idempotency, etc) you bring more complexity and problems to the table.
zug_zug 7y ago
I LOVE this idea. I usually hear other Sr. engineers denigrate it as "hacky," but I think they aren't really looking at the big picture. 1. By combining services, 1 less service to manage in your stack (e.g. do your demo/local/qa envs all connect to Sqs?) 2. Postgres preserves your data if it goes down 3. You already have the tools on each machine and everybody knows the querying language to examine the stack 4. All your existing DB tools (e.g. backup solutions) automatically now cover your queue too, for free. 5. Performance is a non-issue for any company doing < 10m queue items a day.
- KirinDave 7y ago
  > I LOVE this idea. I usually hear other Sr. engineers denigrate it as "hacky," but I think they aren't really looking at the big picture. Because it is hacky from the perspective of a distributed system architecture. It's coupling 2 components that probably ought not be coupled because it's perceived as "convenient" to do so. The idea that your system's control and data planes are tightly coupled is a dangerous one if your system grows quickly. It forces a lot of complexity into areas where you'd rather not deal with it, later manifesting as technical debt as folks who need to solve the problem have to go hunting through the codebase to find where all the control and throttling decisions are being made and what they touch. > 1. By combining services, 1 less service to manage in your stack (e.g. do your demo/local/qa envs all connect to Sqs?) "Oh this SQS is such a pain to manage", is not something anyone who has used SQS is going to say. It's much less work than a database. It has other benefits too, like being easier for devs by being able to trivially replicate staging and production environments is a major selling point of cloud services in the first place. And unlike Postgres, you can do it from anywhere. The decision to use the same postgres you use for data as a queue also concentrates failure in your app and increases the blast radius of performance issues concentrated around one set of hardware interfaces. A bad day for your disk can now manifest in many more ways. > 2. Postgres preserves your data if it goes down Not if you're using a SKIP LOCKED queue (in the sense that it has exactly the same failure semantics as SQS, and you're more likely to lose your postgres instance than SQS with a whole big SRE team is to go down). Can you name a data loss scenario for SQS in the past 8 years? I can think of one that didn't hit every customer, and I'm not at AWS. Personally, I haven't lost data to it despite a fair amount of use since the time us-east-1 flooded. Certainly "reliability" isn't SQS's primary problem. > 3. You already have the tools on each machine and everybody knows the querying language to examine the stack SQS has a web console. You don't need to know a querying language. If you're querying off a queue, you have ceased to have an unqualified queue. > 4. All your existing DB tools (e.g. backup solutions) automatically now cover your queue too, for free. Given most folks who are using AWS are strongly encouraged and empoewred to use RDS, this seems like a wash. But also: SHOULD you back up your queues? This seems to me like an important architectural question. If your system goes down hard, should/can it resume where it left off when restarting? It'd argue the answer is almost always "no." But YMMV. > 5. Performance is a non-issue for any company doing < 10m queue items a day. You say, and then your marketing team says, "We're gonna be on Ellen next week, get ready!"
andrewstuart 7y ago
Here is a complete implementation: 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()
- _sh 7y ago
  Interesting... How do you schedule this? If the queue is empty, do you back off and retry later, or spin the query until it returns a queue item, or some other way? It's a nice approach.
- skc 7y ago
  Thank you for posting this. Will definitely be using it going forward. Beautifully simple.

varelaz 7y ago

There are VPC API endpoints for this. As far as I recall it was all the time there from VPC begging, but I might be wrong.

andrewstuart 7y ago
Only since December 2018 and you need to pay extra... https://aws.amazon.com/about-aws/whats-new/2018/12/amazon-sqs-vpc-endpoints-aws-privatelink/ https://aws.amazon.com/about-aws/whats-new/2018/12/amazon-sq...

LunaSea 7y ago

Do you have an example of such a queue somewhere? I think I have a rough idea about how it works because I implemented something similar about four years ago in PostgreSQL but kept getting locking issues I couldn't get out of: - https://stackoverflow.com/questions/33467813/concurrent-crawler-postgresql-query-issues https://stackoverflow.com/questions/33467813/concurrent-craw... - https://stackoverflow.com/questions/29807033/postgresql-concurrent-transaction-issues https://stackoverflow.com/questions/29807033/postgresql-conc... Also, what kind of queue size / concurrency on the polling side are you able to sustain for your current hardware?

sorokod 7y ago
Here https://www.2ndquadrant.com/en/blog/what-is-select-skip-locked-for-in-postgresql-9-5/ https://www.2ndquadrant.com/en/blog/what-is-select-skip-lock... is a good discussion.
andrewstuart 7y ago
Refer example code above. I don't think Postgres included SKIP LOCKED 4 years ago.
floatboth 7y ago
There's a few libraries implementing this: https://github.com/mbuhot/ecto_job https://github.com/mbuhot/ecto_job for Elixir https://github.com/timgit/pg-boss https://github.com/timgit/pg-boss for Node.js

sorokod 7y ago

Using SKIP LOCKED - do you commit the change to the dequeued item (ack it) at the point where you exit the DB call. If so what happens if the instance that dequeued the messages crashes?

napsterbr 7y ago
Not GP but I think this wouldn't be a problem. The consumer dequeues with SELECT FOR UPDATE within a transaction. If it crashes the database would rollback the transaction, and then another consumer would be able to select the work unit. As for acking, I see two common methods: using an additional boolean column, something like is_processed. Consumers skip truthy ones. Or, after the work is done, simply delete the entry or move elsewhere (e.g. For archival / auditing).
- sorokod 7y ago
  My question assumed a scenario where a consumer dequeues a batch, commits the deqieued change, and then crashes while processing the batch. Offcourse one could delay the commit until all processing is completed but then reasoning about the queue throughput becomes tricky.
  - napsterbr 7y ago
    That's the challenge of distributed systems :) it really boils down to how you want failures to be handled. If you ack before processing, and then you crash, those messages are lost (assuming you can't recover from the crash and you are not using something like a two-phase commit). If you ack after processing, you may fail after the messages have been processed but before you've been able to ack them. This leads to duplicates, in which case you better hope your work units are idempotent. If they are not, you can always keep a separate table of message IDs that have been processed, and check against it. Either way, it's hard, complex and there are thousands of intermediate failure cases you have to think about. And for each possible solution (2pc, separate table of message IDs for idempotency, etc) you bring more complexity and problems to the table.

zug_zug 7y ago

I LOVE this idea. I usually hear other Sr. engineers denigrate it as "hacky," but I think they aren't really looking at the big picture. 1. By combining services, 1 less service to manage in your stack (e.g. do your demo/local/qa envs all connect to Sqs?) 2. Postgres preserves your data if it goes down 3. You already have the tools on each machine and everybody knows the querying language to examine the stack 4. All your existing DB tools (e.g. backup solutions) automatically now cover your queue too, for free. 5. Performance is a non-issue for any company doing < 10m queue items a day.

KirinDave 7y ago
> I LOVE this idea. I usually hear other Sr. engineers denigrate it as "hacky," but I think they aren't really looking at the big picture. Because it is hacky from the perspective of a distributed system architecture. It's coupling 2 components that probably ought not be coupled because it's perceived as "convenient" to do so. The idea that your system's control and data planes are tightly coupled is a dangerous one if your system grows quickly. It forces a lot of complexity into areas where you'd rather not deal with it, later manifesting as technical debt as folks who need to solve the problem have to go hunting through the codebase to find where all the control and throttling decisions are being made and what they touch. > 1. By combining services, 1 less service to manage in your stack (e.g. do your demo/local/qa envs all connect to Sqs?) "Oh this SQS is such a pain to manage", is not something anyone who has used SQS is going to say. It's much less work than a database. It has other benefits too, like being easier for devs by being able to trivially replicate staging and production environments is a major selling point of cloud services in the first place. And unlike Postgres, you can do it from anywhere. The decision to use the same postgres you use for data as a queue also concentrates failure in your app and increases the blast radius of performance issues concentrated around one set of hardware interfaces. A bad day for your disk can now manifest in many more ways. > 2. Postgres preserves your data if it goes down Not if you're using a SKIP LOCKED queue (in the sense that it has exactly the same failure semantics as SQS, and you're more likely to lose your postgres instance than SQS with a whole big SRE team is to go down). Can you name a data loss scenario for SQS in the past 8 years? I can think of one that didn't hit every customer, and I'm not at AWS. Personally, I haven't lost data to it despite a fair amount of use since the time us-east-1 flooded. Certainly "reliability" isn't SQS's primary problem. > 3. You already have the tools on each machine and everybody knows the querying language to examine the stack SQS has a web console. You don't need to know a querying language. If you're querying off a queue, you have ceased to have an unqualified queue. > 4. All your existing DB tools (e.g. backup solutions) automatically now cover your queue too, for free. Given most folks who are using AWS are strongly encouraged and empoewred to use RDS, this seems like a wash. But also: SHOULD you back up your queues? This seems to me like an important architectural question. If your system goes down hard, should/can it resume where it left off when restarting? It'd argue the answer is almost always "no." But YMMV. > 5. Performance is a non-issue for any company doing < 10m queue items a day. You say, and then your marketing team says, "We're gonna be on Ellen next week, get ready!"

andrewstuart 7y ago

Here is a complete implementation: 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()

_sh 7y ago
Interesting... How do you schedule this? If the queue is empty, do you back off and retry later, or spin the query until it returns a queue item, or some other way? It's a nice approach.
skc 7y ago
Thank you for posting this. Will definitely be using it going forward. Beautifully simple.