fix: call pg_notification_queue_usage() automatically when the LISTEN channel bug is detected (#4858)

There's a PostgreSQL bug that doesn't let any listener to register in the DB:
https://www.postgresql.org/message-id/flat/CAK98qZ3wZLE-RZJN_Y%2BTFjiTRPPFPBwNBpBi5K5CU8hUHkzDpw%40mail.gmail.com
The only workaround is to advance the async notification queue tail,
which can be done by executing:  "SELECT pg_notification_queue_usage();".
Before we just logged a HINT with this suggestion, but now we call that function directly
and then let the listener to automatically recover.

No automated tests were added here as it would be too complex and this is a PostgreSQL bug.
But this was manually tested following the steps on
#4581 (comment)

Author: laurenceisla

CHANGELOG.md

@@ -8,6 +8,7 @@ All notable changes to this project will be documented in this file. From versio
 
 - Fix login with uppercase and mixed case role names by @taimoorzaeem in #4678
 - Restore Listener query shape so it can be found in pg_stat_activity by @mkleczek in #4857 #4859
+- The LISTEN channel now automatically recovers when it stops working due to a PostgreSQL bug @laurenceisla in #3147
 
 ## [14.10] - 2026-04-16
 

src/PostgREST/Listener.hs

@@ -10,8 +10,7 @@ import qualified Hasql.Connection      as SQL
 import qualified Hasql.Notifications   as SQL
 import           PostgREST.AppState    (AppState, getConfig)
 import           PostgREST.Config      (AppConfig (..))
-import           PostgREST.Observation (Observation (..),
-                                        isDbListenerBug)
+import           PostgREST.Observation (Observation (..))
 import           PostgREST.Version     (prettyVersion)
 
 import qualified PostgREST.AppState as AppState
@@ -20,6 +19,7 @@ import qualified PostgREST.Config   as Config
 import           Control.Arrow              ((&&&))
 import           Data.Bitraversable         (bisequence)
 import           Data.Either.Combinators    (whenRight)
+import qualified Data.Text                  as T
 import qualified Database.PostgreSQL.LibPQ  as LibPQ
 import qualified Hasql.Session              as SQL
 import           PostgREST.Config.Database  (queryPgVersion)
@@ -31,20 +31,20 @@ runListener :: AppState -> IO ()
 runListener appState = do
   AppConfig{..} <- getConfig appState
   when configDbChannelEnabled $
-    void . forkIO . void $ retryingListen appState
+    void . forkIO . void $ retryingListen appState False
 
 -- | Starts a LISTEN connection and handles notifications. It recovers with exponential backoff with a cap of 32 seconds, if the LISTEN connection is lost.
 -- | This function never returns (but can throw) and return type enforces that.
-retryingListen :: AppState -> IO Void
-retryingListen appState = do
+retryingListen :: AppState -> Bool -> IO Void
+retryingListen appState hasDbListenerBug = do
   AppConfig{..} <- AppState.getConfig appState
   let
     dbChannel = toS configDbChannel
     onError err = do
       AppState.putIsListenerOn appState False
       observer $ DBListenFail dbChannel (Right err)
       when (isDbListenerBug err) $
-        observer DBListenBugHint
+        observer DBListenBugCallQueryFix
       unless configDbPoolAutomaticRecovery $
         killThread mainThreadId
 
@@ -55,7 +55,7 @@ retryingListen appState = do
       unless (delay == maxDelay) $
         AppState.putNextListenerDelay appState (delay * 2)
       -- loop running the listener
-      retryingListen appState
+      retryingListen appState (isDbListenerBug err)
 
   -- Execute the listener with with error handling
   handle onError $ do
@@ -70,6 +70,7 @@ retryingListen appState = do
         Right db -> do
           (pqHost, pqPort) <- SQL.withLibPQConnection db $ bisequence . (LibPQ.host &&& LibPQ.port)
           pgFullName <- SQL.run (queryPgVersion False) db >>= either throwIO (pure . pgvFullName)
+          when hasDbListenerBug $ SQL.run callNotifQueryUsage db >>= either throwIO pure
           SQL.listen db $ SQL.toPgIdentifier dbChannel
 
           AppState.putIsListenerOn appState True
@@ -106,3 +107,10 @@ retryingListen appState = do
       AppState.schemaCacheLoader appState
 
     releaseConnection = void . forkIO . handle (observer . DBListenerConnectionCleanupFail) . SQL.release
+
+    isDbListenerBug e = "could not access status of transaction" `T.isInfixOf` show e
+
+    -- Used to fix a Postgres bug in the listener, see: https://github.com/PostgREST/postgrest/issues/3147#issuecomment-3494591361
+    -- This query advances the async notification query tail, which solves this issue.
+    callNotifQueryUsage :: SQL.Session ()
+    callNotifQueryUsage = SQL.sql "SELECT pg_notification_queue_usage();"

src/PostgREST/Observation.hs

@@ -13,7 +13,6 @@ module PostgREST.Observation
   , observationMessage
   , ObservationHandler
   , showOnSingleLine
-  , isDbListenerBug
   ) where
 
 import qualified Data.ByteString.Lazy       as LBS
@@ -48,7 +47,7 @@ data Observation
   | DBListenStart (Maybe ByteString) (Maybe ByteString) Text Text -- host, port, version string, channel
   | DBListenFail Text (Either SQL.ConnectionError SomeException)
   | DBListenRetry Int
-  | DBListenBugHint -- https://github.com/PostgREST/postgrest/issues/3147
+  | DBListenBugCallQueryFix
   | DBListenerGotSCacheMsg ByteString
   | DBListenerGotConfigMsg ByteString
   | DBListenerConnectionCleanupFail SomeException
@@ -122,8 +121,8 @@ observationMessage = \case
       either showListenerConnError showListenerException listenErr
   DBListenRetry delay ->
     "Retrying listening for database notifications in " <> (show delay::Text) <> " seconds..."
-  DBListenBugHint ->
-    "HINT:  This is likely a bug in the notification queue, try executing the following to solve it: select pg_notification_queue_usage();"
+  DBListenBugCallQueryFix ->
+    "This is likely a PostgreSQL bug in the notification queue, executing the following to try to solve it: SELECT pg_notification_queue_usage();"
   DBListenerGotSCacheMsg channel ->
     "Received a schema cache reload message on the " <> show channel <> " channel"
   DBListenerGotConfigMsg channel ->
@@ -188,6 +187,3 @@ observationMessage = \case
 
 showOnSingleLine :: Char -> Text -> Text
 showOnSingleLine split txt = T.intercalate " " $ T.filter (/= split) <$> T.lines txt -- the errors from hasql-notifications come intercalated with "\t\n"
-
-isDbListenerBug :: SomeException -> Bool
-isDbListenerBug e = "could not access status of transaction" `T.isInfixOf` show e