Postgres Queue

[Jaymon]

• Choose Postgres queue technology
• Hacker News

[Jaymon]

https://github.com/Jaymon/morp/issues/12

[Jaymon]

Tabs and searches I had open while working on this:

• Devious SQL: Message Queuing Using Native PostgreSQL - this was one of the most valuable posts for helping me understand how everything works on the Postgres side.
• Implementing Queues in PostgreSQL (Part 1: Design and Measurement)
• https://github.com/QueueClassic/queue_classic
• https://github.com/gavinwahl/django-postgres-queue
• Most work queue implementations in SQL are wrong
• Do you really need Redis? How to get away with just PostgreSQL
• Solid Queue & understanding UPDATE SKIP LOCKED
• skip locked update instead of delete for queue
• Wait until row becomes available with SKIP LOCKED
• postgres select and wait until timeout if no results
• Streaming Updates from Postgres
• https://news.ycombinator.com/item?id=12664234
• https://github.com/acaloiaro/neoq/blob/main/backends/postgres/postgres_backend.go
• Postgres: a Better Message Queue than Kafka? (via)
• postgres queue
• postgres select for update

This is the post that got me started looking into this, it included this snippet of code which was a major inspiration on getting me to actually implement this:

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()