MIME-Version: 1.0
From: Alex Burkhart <burkharta77@gmail.com>
Date: Sat, 18 Jan 2025 19:44:07 +0900
Message-ID: <CA+vxVFNLn7TOOmE+DRfOJ2z-fpaHYveRg9mq+tYZYG_pckBMfA@mail.gmail.com>
Subject: Design of a reliable task processing queue
To: pgsql-general@lists.postgresql.org
Content-Type: multipart/alternative; boundary="000000000000adbffc062bf8b560"
Archived-At: <https://www.postgresql.org/message-id/CA%2BvxVFNLn7TOOmE%2BDRfOJ2z-fpaHYveRg9mq%2BtYZYG_pckBMfA%40mail.gmail.com>
Precedence: bulk

--000000000000adbffc062bf8b560
Content-Type: text/plain; charset="UTF-8"

Hey team,

I'm looking for help to organize locks and transaction for a reliable task
queue.

REQUIREMENTS

1. Pending actions are persisted to a database. There's a trace once they
are done.
2. Application workers pick actions one by one. At any given time, each
action can be assigned to at most one worker (transaction).
3. If multiple actions have same "lock_id", only one of them is processed
at the time. That has to be action with smallest id.

MY ATTEMPT

I got something approximate working with the following setup.

=== BEGIN DATABASE QUERY ===

DROP TABLE IF EXISTS actions;

CREATE TABLE actions (
  id SERIAL PRIMARY KEY,
  lock_id BIGINT,
  action VARCHAR(255),
  done BOOLEAN DEFAULT false
);

-- Sample data for testing.

INSERT INTO actions (lock_id, action) VALUES
  (26, 'Create instance 26'),
  (8, 'Update instance 8'),
  (26, 'Update instance 26'),
  (8, 'Delete instance 8');

=== END DATABASE QUERY ===

I use Go client to simulate workers utilizing the "actions" table.

=== BEGIN GO CLIENT ===

package main

import (
"context"
"fmt"
"os"
"os/signal"
"time"

"github.com/jackc/pgx/v5/pgxpool"
)

func main() {
ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
defer cancel()

pool, err := pgxpool.New(ctx, "postgres://postgres:password@localhost
/database")
if err != nil {
panic(err)
}
defer pool.Close()

tx, err := pool.Begin(ctx)
if err != nil {
panic(err)
}
defer tx.Rollback(ctx)

fmt.Println("BEGIN")

var (
id     int32
lockid int64
action string
)
if err := tx.QueryRow(
ctx,
`SELECT id, lock_id, action
     FROM actions
     WHERE done = false
     ORDER BY id
     LIMIT 1
     FOR NO KEY UPDATE
     SKIP LOCKED`,
).Scan(&id, &lockid, &action); err != nil {
panic(err)
}

fmt.Println("LOCKING", lockid, "...")

if _, err := tx.Exec(ctx, fmt.Sprintf("SELECT pg_advisory_xact_lock(%d)",
lockid)); err != nil {
panic(err)
}

fmt.Println("EXECUTING", action)

select {
case <-ctx.Done():
// Pretend to do work for 10 seconds.
case <-time.After(10 * time.Second):
if _, err := tx.Exec(ctx, "UPDATE actions SET done = true WHERE id = $1",
id); err != nil {
panic(err)
}
if err := tx.Commit(ctx); err != nil {
panic(err)
}

fmt.Println("DONE")
}

fmt.Println("UNLOCKED")
}

=== END GO CLIENT ===

This code generates transaction like this.

=== BEGIN TRANSACTION ===

BEGIN;

-- Lock one row in "actions" table.

SELECT id, lock_id, action
FROM actions
WHERE done = false
ORDER BY id
LIMIT 1
FOR NO KEY UPDATE
SKIP LOCKED;

-- Lock other transactions that process same lock_id.

SELECT pg_advisory_xact_lock(%lock_id);

-- Work on the action... mark it done at the end.

UPDATE actions SET done = true WHERE id = %d;

COMMIT;

=== END TRANSACTION ===

Which almost does the job. By running the Go client concurrently, it picks
actions one by one and processes only one transaction with same "lock_id"
at a time.

However I'm worried about the gap between the row lock and "lock_id" lock.
This leaves a room for requirement "That has to be action with smallest id"
to be unsatisfied.

QUESTION

Is there a way to improve this attempt and close the gap? Or a completely
different strategy? I was brainstorming how to lock all rows where columns
have the same value or using ARRAY but struggle to put together a reliable
solution.

Thank you,
Alex

--000000000000adbffc062bf8b560
Content-Type: text/html; charset="UTF-8"
Content-Transfer-Encoding: quoted-printable

<div dir=3D"ltr">Hey team,<br><br>I&#39;m looking for help to organize lock=
s and transaction for a reliable task queue.<br><br>REQUIREMENTS<br><br>1. =
Pending actions are persisted to a database. There&#39;s a trace once they =
are done.<br>2. Application workers pick actions one by one. At any given t=
ime, each action can be assigned to at most one worker (transaction).<br>3.=
 If multiple actions have same &quot;lock_id&quot;, only one of them is pro=
cessed at the time. That has to be action with smallest id.<br><br>MY ATTEM=
PT<br><br>I got something approximate working with the following setup.<br>=
<br>=3D=3D=3D BEGIN DATABASE QUERY =3D=3D=3D<br><br>DROP TABLE IF EXISTS ac=
tions;<br><br>CREATE TABLE actions (<br>=C2=A0 id SERIAL PRIMARY KEY,<br>=
=C2=A0 lock_id BIGINT,<br>=C2=A0 action VARCHAR(255),<br>=C2=A0 done BOOLEA=
N DEFAULT false<br>);<br><br>-- Sample data for testing.<br><br>INSERT INTO=
 actions (lock_id, action) VALUES<br>=C2=A0 (26, &#39;Create instance 26=
9;),<br>=C2=A0 (8, &#39;Update instance 8&#39;),<br>=C2=A0 (26, &#39;Update=
 instance 26&#39;),<br>=C2=A0 (8, &#39;Delete instance 8&#39;);<br><br>=3D=
=3D=3D END DATABASE QUERY =3D=3D=3D<br><br>I use Go client to simulate work=
ers utilizing the &quot;actions&quot; table.<br><br>=3D=3D=3D BEGIN GO CLIE=
NT =3D=3D=3D<br><br>package main<br><br>import (<br>&quot;context&quot;<br>=
&quot;fmt&quot;<br>&quot;os&quot;<br>&quot;os/signal&quot;<br>&quot;time&qu=
ot;<br><br>&quot;<a href=3D"http://github.com/jackc/pgx/v5/pgxpool">github.=
com/jackc/pgx/v5/pgxpool</a>&quot;<br>)<br><br>func main() {<br>ctx, cancel=
 :=3D signal.NotifyContext(context.Background(), os.Interrupt)<br>defer can=
cel()<br><br>pool, err :=3D pgxpool.New(ctx, &quot;postgres://postgres:pass=
word@localhost/database&quot;)<br>if err !=3D nil {<br>panic(err)<br>}<br>d=
efer pool.Close()<br><br>tx, err :=3D pool.Begin(ctx)<br>if err !=3D nil {<=
br>panic(err)<br>}<br>defer tx.Rollback(ctx)<br><br>fmt.Println(&quot;BEGIN=
&quot;)<br><br>var (<br>id =C2=A0 =C2=A0 int32<br>lockid int64<br>action st=
ring<br>)<br>if err :=3D tx.QueryRow(<br>ctx,<br>`SELECT id, lock_id, actio=
n<br>=C2=A0 =C2=A0 =C2=A0FROM actions<br>=C2=A0 =C2=A0 =C2=A0WHERE done =3D=
 false<br>=C2=A0 =C2=A0 =C2=A0ORDER BY id<br>=C2=A0 =C2=A0 =C2=A0LIMIT 1<br=
>=C2=A0 =C2=A0 =C2=A0FOR NO KEY UPDATE<br>=C2=A0 =C2=A0 =C2=A0SKIP LOCKED`,=
<br>).Scan(&amp;id, &amp;lockid, &amp;action); err !=3D nil {<br>panic(err)=
<br>}<br><br>fmt.Println(&quot;LOCKING&quot;, lockid, &quot;...&quot;)<br><=
br>if _, err :=3D tx.Exec(ctx, fmt.Sprintf(&quot;SELECT pg_advisory_xact_lo=
ck(%d)&quot;, lockid)); err !=3D nil {<br>panic(err)<br>}<br><br>fmt.Printl=
n(&quot;EXECUTING&quot;, action)<br><br>select {<br>case &lt;-ctx.Done():<b=
r>// Pretend to do work for 10 seconds.<br>case &lt;-time.After(10 * time.S=
econd):<br>if _, err :=3D tx.Exec(ctx, &quot;UPDATE actions SET done =3D tr=
ue WHERE id =3D $1&quot;, id); err !=3D nil {<br>panic(err)<br>}<br>if err =
:=3D tx.Commit(ctx); err !=3D nil {<br>panic(err)<br>}<br><br>fmt.Println(&=
quot;DONE&quot;)<br>}<br><br>fmt.Println(&quot;UNLOCKED&quot;)<br>}<br><br>=
=3D=3D=3D END GO CLIENT =3D=3D=3D<br><br>This code generates transaction li=
ke this.<br><br>=3D=3D=3D BEGIN TRANSACTION =3D=3D=3D<br><br>BEGIN;<br><br>=
-- Lock one row in &quot;actions&quot; table.<br><br>SELECT id, lock_id, ac=
tion<br>FROM actions<br>WHERE done =3D false<br>ORDER BY id<br>LIMIT 1<br>F=
OR NO KEY UPDATE<br>SKIP LOCKED;<br><br>-- Lock other transactions that pro=
cess same lock_id.<br><br>SELECT pg_advisory_xact_lock(%lock_id);<br><br>--=
 Work on the action... mark it done at the end.<br><br>UPDATE actions SET d=
one =3D true WHERE id =3D %d;<br><br>COMMIT;<br><br>=3D=3D=3D END TRANSACTI=
ON =3D=3D=3D<br><br>Which almost does the job. By running the Go client con=
currently, it picks actions one by one and processes only one transaction w=
ith same &quot;lock_id&quot; at a time.<br><br>However I&#39;m worried abou=
t the gap between the row lock and &quot;lock_id&quot; lock. This leaves a =
room for requirement &quot;That has to be action with smallest id&quot; to =
be unsatisfied.<br><br>QUESTION<br><br>Is there a way to improve this attem=
pt and close the gap? Or a completely different strategy? I was brainstormi=
ng how to lock all rows where columns have the same value or using ARRAY bu=
t struggle to put together a reliable solution.<br><br>Thank you,<br>Alex</=
div>

--000000000000adbffc062bf8b560--