How/why to sweep async tasks under a Postgres table
taylor.town
taylor.town
> Never Handroll Your Own Two-Phase Commit
And then buried at the end:
> A few notable features of this snippet:
> Limiting number of retries makes the code more complicated, but definitely worth it for user-facing side-effects like emails.
This isn't two-phase commit. This is lock the DB indefinitely while remote system is processing and pray we don't crash saving the transaction after it completes. That locked also eats up a database connection so your concurrency is limited by the size of your DB pool.
More importantly, if the email sends but the transaction to update the task status fails, it will try again. And again. Forever. If you're going to track retries it would have to be before you start the attempt. Otherwise the "update the attempts count" logic itself could fail and lead to more retries.
The real answer to all this is to use a provider that supports idempotency keys. Then when you can retry the action repeatedly without it actually happening again. My favorite article on this subject: https://brandur.org/idempotency-keys
I agree the concurrency limitation is kind of rough, but it's kind of elegant because you don't have to implement some kind of timeout/retry thing. You're certainly still exposed to the possibility of double-sending, so yes, probably much nicer to update the row to "processing" and re-process those rows on a timeout.
It was outside of the scope of this essay, but a lot of these problems can be resolved with a mid-transaction COMMIT and reasonable timeouts
You can implement a lean idempotency system within the task pattern like this, but it really depends on what you need and what failures you want to prevent
Thanks for providing more context and safety tips! :)
One reason to do this for emails, IE a database queue is to keep a log/history of all sent emails, as well as a render for "view in browser" links in the email itself. Not to mention those rare instances where an email platform goes down and everything starts blowing up.
Turtles all the way down?
Let's say you are the provider that must support idempotency keys? How should it be done?
Agreed
> This is lock the DB indefinitely while remote system is processing and pray we don't crash saving the transaction after it completes.
I don't really see the problem here (maybe due to my node.js skillz being less than excellent), because I don't see how it's locking the table; that one row would get locked, though.
I prefer an optimistic locking solution for this type of thing.
TBH, I'm not too worried about that either - my understanding from the content is that you'd have a few tasks running in the background that service the queue; even one is enough.
I'd just consider that to be always-active, and turn the knobs accordingly.
> It also hurts things like reindexing.
I dunno about this one - does re-indexing on the queue occur often enough to matter? After all, this is a queue of items to process. I'd be surprised if it needed to be re-indexed at all.
But I think it totally depends on what your queue is used for. In my case, I need durable queues that report status/errors and support retries/back off.
I deal with it using updates rather than deleting from the queue, because I need a log of what happened for audit purposes. If I need to optimize later, I can easily partition the table. At the start, I just use a partial index for the items to be processed.
Reindexing, and other maintenance functions that need to rewrite the table will happen more than you like in a production system, so I'd rather make them easy to do.
Agreed
> I deal with it using updates rather than deleting from the queue, because I need a log of what happened for audit purposes. If I need to optimize later, I can easily partition the table. At the start, I just use a partial index for the items to be processed.
That's a good approach. I actually have this type of queue in production, and my need is similar to yours, but the expected load is a lot less - there's an error if the application goes through a day and sees even a few thousand work items added to the queue (this queue is used for user notifications, and even very large clients have only a few thousand users).
So, my approach is to have a retry column that decrements to zero each time I retry a work item, with items having a zero in the retry column getting ignored.
The one worker runs periodically (currently every 1m) and processes only those rows with a non-zero retry column and with the `incomplete` flag set.
A different worker runs every 10m and moves expired rows (those with the retry column set to zero) and completed rows (those with a column set to `done` or similar) to a different table for audit/logging purposes. This is why I said that the table containing workitems will almost never be reindexed: all rows added will eventually be removed.
------------------------------------------------
The real problem is that the processing cannot be done atomically, even when there is only a single worker.
For example, if the processing is "send email", your system might go down after calling the `send_email()` function in your code and before calling the `decrement_retry()` in the code.
No amount of locking, n-phase commits, etc can ever prevent the case where the email is sent but the retry counter is not decremented. This is not a solvable problem so I am prepared to live with it for now with the understanding that the odds are low that this situation will come up, and if it does the impact is lows as well (user gets two notification emails for the same item).
This is much more complicated with task queues. Doable still! But often skipped, because it's tempting to imagine that the backend will just handle the failure by retrying. But there are lots of kinds of failure that can happen.
The recipient's server doesn't accept the email. The recipient's domain name expired. Actually, we don't have an email address for that recipient at all.
The user has seen "got it, will do, don't worry about it" but if that email is time sensitive, they might want to know that it hasn't been sent yet, and maybe they should place a phone call instead.
After all Amazon does this for a physical order. It may be days before a status update!
It does rely on redis, but it's basically the same idea.
Turns out the article advocates exactly that. The example uses CTEs with multi-table inserts. "Dumb" here means "no synchronous external service calls," not "avoid complex SQL."
Fun fact: A query like this will, once in a blue moon, return more than limit (here 1) row, since the inner query is executed multiple times and returns different ids, which is surprising for a lot of people. If your code does not expect that, it may cause problems. (The article seems to, since it uses a list and iteration to handle the result.)
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
Also, if your tasks take a long time, it will create long-running transactions, which may cause dead tuple problems. If you need to avoid that, you can mark the task as "running" in a short-lived transaction and delete it in another. It becomes more complicated then, since you need to handle the case that your application dies while it has taken a task.
It looks like multiple task rows are being retrieved via a delete...returning statement, and for each row an email being sent. If there's an error, the delete statement is rolled back.
Let's hypothesize that a batch of ten tasks are retrieved, and the 9th has a bad email address, so the batch gets rolled back on error. Next retry the welcome email would be sent again for the ones that succeeded, right?
Even marking the task as "processed" with the tx in the task code wouldn't work, because that update statement would also get rolled back.
Am I missing something? (entirely possible, the code is written in "clever" style that makes it harder for me to understand, especially the bit where $sql(usr_id) is passed into the sql template before it's been returned, is there CTE magic there that I don't understand?)
I thought that this was the reason that most systems like this only pull one row at a time from the queue with skip locked...
Thanks to anyone who can clarify this for me if it is indeed correct!
It's a confusing way to do do things to me, like, why not select ordered by task date limit 1? Still using for update and skip locked etc... hold the transaction, and update to 'complete' or delete/move the row when done? What's the advantage of the inner select like that?
And I'm still totally confused by:
const [{ usr_id } = { usr_id: null }] = await sql`
with usr_ as (
insert into usr (email, password)
values (${email}, crypt(${password}, gen_salt('bf')))
returning *
), task_ as (
insert into task (task_type, params)
values ('SEND_EMAIL_WELCOME', ${sql({ usr_id })})
)
select * from usr_
`;
I think the idea is great, I think I'm just struggling a bit with the code style, it seems to be "clever" in a way that increases cognitive load without a real benefit that I can see, but I suppose that's pretty subjective.
https://www.recall.ai/blog/postgres-listen-notify-does-not-s...
I do not agree that it is complex but that’s what I’d hypothesize as why someone would think that.
I thought I was alone in this, but I build all my personal projects this way! I wish I could use this approach at work, but too many colleagues crave "engineering."
But then again, there's GraphQL because frontend developers thought backend developers are anti social.
If I understand you correctly, I don't think of it as overdoing HTTP verbs so much as using an excessively naive mapping between HTTP resources and base table entities.