Connection.notifies() generator loses notifications

[codeech]

psycopg version: 3.2.3
postgres version: PostgreSQL 14.13 (Homebrew) on x86_64-apple-darwin23.6.0, compiled by Apple clang version 16.0.0 (clang-1600.0.26.4), 64-bit

I have a long-running process (possibly months) that listens for notifications on a database connection using conn.notifies() (see psycopg3 documentation). The documentation doesn't specify the expected behavior if the listening connection becomes unresponsive/corrupt, so I believe I need to perform health checks myself.

If conn.notifies() is guaranteed to raise an exception if the connection has issues, then the rest of this is moot. Does anyone know?

Going off of this comment on issue #340, I believe the following script should not lose notifications sent while select pg_sleep(3) executes and thus allow me to do health checks without losing messages.

def _listen_for_notifications():
    with psycopg.Connection.connect(CONN_STR, autocommit=True) as conn:
        listen_sql = sql.SQL("LISTEN {}").format(sql.Identifier("some_channel_name"))

        conn.execute(listen_sql)
        gen = conn.notifies(timeout=5)
        print('listening')
        # This prints notifications sent within the five second timeout
        for n in gen:
            print(n)

        gen.close()  # I see the same behavior with and without this line.

        print("Performing health check")
        # Any notifications sent over these three seconds are lost
        conn.execute("select pg_sleep(3)")  # Would be SELECT 1 in real code

        conn.execute(listen_sql)
        gen = conn.notifies(timeout=5)
        print('listening again on the same connection')
        # This prints notifications sent within the five second timeout
        for n in gen:
            print(n)

        print('done!')


_listen_for_notifications()

Output when I send a notification with payload '1' during the first 5 seconds, a notification with payload '2' during the 3 seconds pg_sleep(3) executes, and a notification with payload '3' during the last 5 seconds.

listening
Notify(channel='some_channel_name', payload='1', pid=35237)
Performing health check
listening again on the same connection
Notify(channel='some_channel_name', payload='3', pid=35237)
done!

Notification 2 is lost.

[dvarrazzo]

I did some test on my side, and I confirm what you see. My test (to be added in test_notify_async.py) is:

@pytest.mark.slow
@pytest.mark.timing
@pytest.mark.parametrize("sleep_on", ["server", "client"])
async def test_notify_query_notify(aconn_cls, dsn, sleep_on):
    e = AEvent()
    by_gen = []
    by_cb = []
    workers = []

    async def notifier():
        async with await aconn_cls.connect(dsn, autocommit=True) as aconn:
            for i in range(6):
                await aconn.execute("select pg_notify('counter', %s)", (str(i),))
                await asleep(0.5)

    async def listener():
        async with await aconn_cls.connect(dsn, autocommit=True) as aconn:
            aconn.add_notify_handler(lambda n: by_cb.append(int(n.payload)))
            await aconn.execute("listen counter")
            e.set()
            async for n in aconn.notifies(timeout=1):
                by_gen.append(int(n.payload))
            if sleep_on == "server":
                await aconn.execute("select pg_sleep(1)")
            elif sleep_on == "client":
                await asleep(1)
            else:
                assert False, sleep_on
            async for n in aconn.notifies(timeout=2):
                by_gen.append(int(n.payload))

    workers.append(spawn(listener))
    await e.wait()
    workers.append(spawn(notifier))
    await gather(*workers)

    assert list(range(6)) == by_cb == by_gen, f"{by_gen=}, {by_cb=}"

One task notifies every 0.5s with an increasing number as payload, the other task tries to read the notifcations. It does in two ways: by notifies() method and by notifies callback. The expected result is a range(6) list. The listener will listen for one second, wait for one second, listen for another second (two for extra measure).

Result: none of the list is complete 🙁 For the moment I'd focus on the issue seen waiting on the server-side, waiting on the client side is easier to fix: the result is by_gen containing [0, 1, 4, 5], by_cb containing [2, 3], which is kinda fun...

So here I see a bug that is not the one you mention: during notifies(), the notify handlers don't work. This is an easy to fix issue:

--- a/psycopg/psycopg/generators.py
+++ b/psycopg/psycopg/generators.py
@@ -297,6 +297,8 @@ def notifies(pgconn: PGconn) -> PQGen[list[pq.PGnotify]]:
     while True:
         n = pgconn.notifies()
         if n:
+            if pgconn.notify_handler:
+                pgconn.notify_handler(n)
             ns.append(n)
         else:
             break

With this change, the sleep_on == "client" test passes, the sleep_on == "server" test still fails, but only by_gen will miss the two notifications.

So, the most ready conclusion that we can draw is that my comment in #340 was wrong. The reason is that yes, the libpq stores the notifications in the connection state, but, in the current state, _consume_notifes() will just plonk them.

psycopg/psycopg/psycopg/generators.py

Lines 282 to 289 in c7b8ed5

	def _consume_notifies(pgconn: PGconn) -> None:
	# Consume notifies
	while True:
	n = pgconn.notifies()
	if not n:
	break
	if pgconn.notify_handler:
	pgconn.notify_handler(n)

I don't know if this behaviour got changed in the time since that comment was made, or if I've been wrong all the time. For sure that comment at line 283 ranks among the least informative comments in humanity 😬

So, accepting the comment being wrong, we remain with the issue of allowing the users to write a robust listener that can be interrupted, the connection used to run a query, then listened on again. I see two possibilities: one that you will hate, one that I will hate 😄

1. We don't do anything and we instruct people to use a handler to fill the gap. They would end up having to write code such as:

    async def listener():
        async with await aconn_cls.connect(dsn, autocommit=True) as aconn:
            def patch(n):
                by_gen.append(int(n.payload))

            await aconn.execute("listen counter")

            async for n in aconn.notifies(timeout=1):
                by_gen.append(int(n.payload))

            aconn.add_notify_handler(patch)
            await aconn.execute("select pg_sleep(1)")
            aconn.remove_notify_handler(patch)

            async for n in aconn.notifies(timeout=2):
                by_gen.append(int(n.payload))

you don't have to tell me how much it would suck. Therefore,

2. I would add the connection an internal list of notifications received, populated by a callback; however I wouldn't register this callback by default; I would only register it the first time that notifies() is used. The aim of this callback would be exactly to fill the gap as in the patch() of the example. When notifies() is invoked a second time it would empty the queue, emit its content, and then go back to its normal usage.

This can become a leak, but only if the user calls notifies() once on a connection and then uses it for a long time, never closing it, never calling notifies() again, and without unlistening the channel. I think that these condition are not too common to worry and most of the times people listening on a connection will want to keep on listening on it, modulo stopping every now and then to run some command to that connection.

What do you think?

[dvarrazzo]

About this:

If conn.notifies() is guaranteed to raise an exception if the connection has issues, then the rest of this is moot. Does anyone know?

If the connection gets closed the kernel should awake the notifier, which should notice the broken connection as soon as it tries to read from it and throw an exception.

However I don't think it's wise to rely on a connection being open for such a long time. You will easily find middleware that may close the connection if they sense it being unused for long. Managed databases might be particularly aggressive in doing so. So maybe it would be wise to set up some form of connection rotation and maybe replace your listening connection every several minutes.

(I personally had a bad experience with haproxy in front of a master+hot standby pair of servers in HA configuration, therefore I try not to keep using the same connection for too long, for example discarding them from pools after about 1/2 the proxy timeout time, and rotating a notifications listener I have in the system after about the same time).

[codeech]

Thank you for the prompt response @dvarrazzo.

This can become a leak, but only if the user calls notifies() once on a connection and then uses it for a long time, never closing it, never calling notifies() again, and without unlistening the channel.

This is my use case. I have two tasks running. One listener, and one worker performing some db operations if a notification comes in. The listener sets an asyncio.Event() which the worker detects and then does the work with a new connection.

I would only want to close the generator to perform a simple "select 1" health check on the listening connection. If the following comment is true, then I would not need to perform these health checks. I could listen indefinitely and only restart the listener if the listening connection has an issue and hence conn.notifies() raises and exception.

If the connection gets closed the kernel should awake the notifier, which should notice the broken connection as soon as it tries to read from it and throw an exception.

This is exactly the behavior I am looking for. Do you know what exception(s) could be thrown specifically? psycopg.errors.OperationalError?

[dvarrazzo]

@codeech I have prepared MR #975 to address the issue you reported. Notifications are now captured whilst the notifies() generator is not listening and returned first.

Feedback is welcome. Cheers!

[dvarrazzo]

Fixed by #975.

[Blackeagle52]

Hi, @dvarrazzo, during a dependency review of the project I'm working on I noticed the new 3.2.4 release, and I saw notifies has been changed. And after reviewing the changes, the changes kinda contradict to the discussion we had in #873.

In my issue I lost Notifications during queries on the same connection between notifies calls. The given example in the description of this issue is quite similar to my issue. As the Notify is being lost after the following query:

conn.execute("select pg_sleep(3)") # Would be SELECT 1 in real code

You stated this as expected behaviour, and as solution you recommended:

if you don't care about the timing of the notification, you can register a handler and handle the notifications both in the poller and anywhere else they happen (you might for instance put them in a queue both in the handler and in the poller and then consume them uniformly).

In my issue you also stated:

you seem to expect the generator will also receive notifications sent before it started. It doesn't work this way; if we wanted so, the connection would need a buffer where to accumulate notifications and then send them out as soon as the generator starts

This accumulator has now actually been added.

As per suggestion, in my app I've combined notifies and add_notify_handler into a working solution. But I never found the energy to finalise my documentation change PR (#878). This PR contained a new example on how to use these together.

But because of #975, using these two methods together now actually lead to duplicate Notify in my queue.

In the end I don't really mind the change, as it allows the removal of the handler. But this change does contradicts to 3.2 stated behaviour, and I would have expected a change as this not to land in a minor update, but in 3.3.

[dvarrazzo]

Hello @Blackeagle52

The conversation we had was instructive for me to understand a genuine problem that users trying to do something relatively straightforward, such as receiving notifications reliably, face. The fix introduced a complexity that originally I would have preferred to avoid, but this would have only meant forcing every single user to implement on their own, which I think would have been a disservice.

So yes, I've backtracked from what I originally stated because I think that the current behaviour is best possible one for everyone. I understand that the correction has caused you a problem (suddenly finding duplicate notifications) and I apologise for it, but I think that having less moving part in your software, and offloading the responsibility of something low-level to the driver rather than the application, is something you will eventually benefit for.

As for when a change should be released according to semver, it's arguable. For me this is a bugfix, we "don't lose notifications anymore" (as many people, likely including you, would initially expect to happen). It doesn't add any interface so it's not a new feature; waiting for the 3.3 release doesn't make sense from semver point of view. The behaviour has slightly changed, which is something that happens at any bugfix (ref. XKCD 1172). So I'm afraid that if this hadn't been released in 3.2.4 it should have waited for psycopg 4.