client side implementation for queue notifications

[ChuckHend]

Notifications were added to pgmq server side in pgmq/pgmq#432. This feature notifies clients when new messages are inserted into the queue tables. We can add an implementation around this to the client side API to make it polling more efficient. There's two cases where continuous polling will still be required:

1. user is sending messages with a VT delay. When using a VT delay, there will be a notificaiton when the message reaches the queue but message will not be immedaitely available.
2. when the message above (or any message) has its VT expire, there is no notificaiton.

There are tests written on the server side project that can be used as a starting point for how to use the notification feature.

[tavallaie]

I’ll work on implementing this. To make sure I fully understand the requirements, here’s my current understanding:

We need two main components:

enable_notifications(queue) and disable_notifications(queue) -> These functions will allow us to start/stop listening for NOTIFICATION events on a specific queue.
read_with_notify(...)
-> This function should wait for a notification before polling.
-> If no notification arrives (e.g., in the case of virtual tables or fallback scenarios), it should gracefully fall back to polling.
Does this sound correct?
@ChuckHend

[ChuckHend]

@tavallaie that sounds right to me. I think it could be nice if read_with_notify() has some sort of parameter that determines the fallback period. For example "poll every 20 seconds, unless there is a notification if which case we read the message(s), then wait 20 seconds before polling again". So its like the current read(), but cuts the interval short when there is a notification.

This is still a bit experimental to IMO, since there are some gotchas related to how we are using NOTIFY (the cases I listed above), so I think we should maybe release as experimental or unstable until we figure out a nice experience around it. @chrisprobst worked on the server side implementation of this feature and might be good include as its built too, and maybe collaborate too.

[chrisprobst]

@ChuckHend Sure, and thanks for creating a separate repo for this based on uv, much cleaner :-).

@tavallaie I've drafted a Python test for this in pgmq/pgmq#431. For completeness, I post this here to discuss the Python side of things:

from queue import Empty, Queue
from threading import Thread

from tembo_pgmq_python.queue import PGMQueue

queue = PGMQueue(
    host="xxx",
    port="5432",
    username="xxx",
    password="xxx",
    database="xxx",
    verbose=True,
    log_filename="pgmq_sync.log",
)


q = Queue(1)


def read_notifies():
    with queue.pool.connection() as conn:
        conn.autocommit = True
        conn.execute("LISTEN tasks;")
        print("Waiting for notifications on channel 'tasks'")
        for notify in conn.notifies():
            print("Got NOTIFY:", notify.pid, notify.channel, notify.payload)
            q.put_nowait(1)


with queue.pool:
    t = Thread(target=read_notifies)
    t.start()
    name = "tasks"
    while True:
        msg = queue.read(name)
        if msg:
            print("Received message", msg)
            assert queue.delete(name, msg.msg_id)
        print("No more messages, waiting for notify or 5 seconds...")
        try:
            q.get(block=True, timeout=5)
            print("Got signal, trying to read ...")
        except Empty:
            print("Timed out, polling again...")
    t.join()

It's only the threaded-version but it already polls regularly and on NOTIFY.

The most important aspect is connection pooling though, that's why I am using a dedicated connection in this example. Calling LISTEN on a pooled conn is obviously problematic because the NOTIFY arrives on exactly that conn. When the conn is released to the pool before, it won't work.

We could also think about calling LISTEN on every conn in the pool and then somehow broadcast this signal to interested threads. Personally, I believe a dedicated connection is better. We can make this more efficient by using a single conn with multiple LISTENs if you have multiple queues and multiplex so that we can support many queues in one conn.

[tavallaie]

@chrisprobst @ChuckHend
Sorry for the late response. I have been working on the implementation, trying to use minimal database connections. However, I have some questions to make sure I understand correctly:

• If there are multiple clients, should all of them receive the NOTIFICATION?
• Should we create multiple connections, or is it better to keep one connection active for getting queues messages and queues NOTIFICATION? And if there are no queue message and no NOTIFICATION, what will happen to the conn? If there is a NOTIFICATION, should we close the conn until the delayed time?
• Should the metrics functions also receive notifications for real-time updates instead of interval updates?
• If we use one connection for multiple queues, how should we handle many notifications coming quickly?

I would like to know your opinion on these matters and what approach you think is better.

[chrisprobst]

@tavallaie Thanks for your thoughts. You are touching on the pain points, you are right.

So, I believe this NOTIFY-based system is not ideal for high throughput scenarios in general. Getting a NOTIFY for every message in a very high throughput scenario would most likely be not efficient.

That being said, I think a single connection, opt-in, to handle all NOTIFYs seems reasonable given the fundamental issue to scale up. Having multiple connections seems even more wasteful. Since it is opt-in, I think leaving it open the entire time is somewhat ok-ish. Opening and closing the conn within delayed time etc. seems also unfeasible due to the overhead new connections create.

This is all by no means perfect. This is useful for this typical "10-100 messages per minute" usecase, probably more. But certainly not in the 1000 or 10K messages / s. To be fair, if you expect that many messages per second, polling with batch read at fixed rates is probably already the best you can do and will most likely outperform even natively designed systems like NATS.

Just my thoughts, hard to decide on right or wrong. I guess even the minimal impl. described above would be useful if latency matters and polling every second is too wasteful.

[tavallaie]

I wrote some draft but I need more time to implement it.