[SOLVED] psycopg3 AsyncConnection connection notifies generator health check

psycopg3 AsyncConnection connection notifies generator health check

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?

Assuming conn.notifies() does not necessarily raise an exception, this github comment says I can execute a statement on the listening connection without losing notifications.

If you need to issue a new LISTEN on that connection, you can stop the notifies generator (using generators.close(), see docs), execute the statement, and call notifies() again. You won't lose messages doing so: they will be buffered in the libpq connection anyway.

def _listen_for_notifications():
    with psycopg.Connection.connect("some_connection_string", 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.

Solution