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 https://github.com/PostgREST/postgrest/pull/4581#issuecomment-3690610592
This commit is contained in:
Laurence Isla
2026-05-04 22:08:06 +05:00
committed by Taimoor Zaeem
parent a62b6de1d0
commit 0502488b70
3 changed files with 19 additions and 14 deletions
+1
View File
@@ -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 - 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 - 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 ## [14.10] - 2026-04-16
+15 -7
View File
@@ -10,8 +10,7 @@ import qualified Hasql.Connection as SQL
import qualified Hasql.Notifications as SQL import qualified Hasql.Notifications as SQL
import PostgREST.AppState (AppState, getConfig) import PostgREST.AppState (AppState, getConfig)
import PostgREST.Config (AppConfig (..)) import PostgREST.Config (AppConfig (..))
import PostgREST.Observation (Observation (..), import PostgREST.Observation (Observation (..))
isDbListenerBug)
import PostgREST.Version (prettyVersion) import PostgREST.Version (prettyVersion)
import qualified PostgREST.AppState as AppState import qualified PostgREST.AppState as AppState
@@ -20,6 +19,7 @@ import qualified PostgREST.Config as Config
import Control.Arrow ((&&&)) import Control.Arrow ((&&&))
import Data.Bitraversable (bisequence) import Data.Bitraversable (bisequence)
import Data.Either.Combinators (whenRight) import Data.Either.Combinators (whenRight)
import qualified Data.Text as T
import qualified Database.PostgreSQL.LibPQ as LibPQ import qualified Database.PostgreSQL.LibPQ as LibPQ
import qualified Hasql.Session as SQL import qualified Hasql.Session as SQL
import PostgREST.Config.Database (queryPgVersion) import PostgREST.Config.Database (queryPgVersion)
@@ -31,12 +31,12 @@ runListener :: AppState -> IO ()
runListener appState = do runListener appState = do
AppConfig{..} <- getConfig appState AppConfig{..} <- getConfig appState
when configDbChannelEnabled $ 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. -- | 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. -- | This function never returns (but can throw) and return type enforces that.
retryingListen :: AppState -> IO Void retryingListen :: AppState -> Bool -> IO Void
retryingListen appState = do retryingListen appState hasDbListenerBug = do
AppConfig{..} <- AppState.getConfig appState AppConfig{..} <- AppState.getConfig appState
let let
dbChannel = toS configDbChannel dbChannel = toS configDbChannel
@@ -44,7 +44,7 @@ retryingListen appState = do
AppState.putIsListenerOn appState False AppState.putIsListenerOn appState False
observer $ DBListenFail dbChannel (Right err) observer $ DBListenFail dbChannel (Right err)
when (isDbListenerBug err) $ when (isDbListenerBug err) $
observer DBListenBugHint observer DBListenBugCallQueryFix
unless configDbPoolAutomaticRecovery $ unless configDbPoolAutomaticRecovery $
killThread mainThreadId killThread mainThreadId
@@ -55,7 +55,7 @@ retryingListen appState = do
unless (delay == maxDelay) $ unless (delay == maxDelay) $
AppState.putNextListenerDelay appState (delay * 2) AppState.putNextListenerDelay appState (delay * 2)
-- loop running the listener -- loop running the listener
retryingListen appState retryingListen appState (isDbListenerBug err)
-- Execute the listener with with error handling -- Execute the listener with with error handling
handle onError $ do handle onError $ do
@@ -70,6 +70,7 @@ retryingListen appState = do
Right db -> do Right db -> do
(pqHost, pqPort) <- SQL.withLibPQConnection db $ bisequence . (LibPQ.host &&& LibPQ.port) (pqHost, pqPort) <- SQL.withLibPQConnection db $ bisequence . (LibPQ.host &&& LibPQ.port)
pgFullName <- SQL.run (queryPgVersion False) db >>= either throwIO (pure . pgvFullName) 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 SQL.listen db $ SQL.toPgIdentifier dbChannel
AppState.putIsListenerOn appState True AppState.putIsListenerOn appState True
@@ -106,3 +107,10 @@ retryingListen appState = do
AppState.schemaCacheLoader appState AppState.schemaCacheLoader appState
releaseConnection = void . forkIO . handle (observer . DBListenerConnectionCleanupFail) . SQL.release 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();"
+3 -7
View File
@@ -13,7 +13,6 @@ module PostgREST.Observation
, observationMessage , observationMessage
, ObservationHandler , ObservationHandler
, showOnSingleLine , showOnSingleLine
, isDbListenerBug
) where ) where
import qualified Data.ByteString.Lazy as LBS 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 | DBListenStart (Maybe ByteString) (Maybe ByteString) Text Text -- host, port, version string, channel
| DBListenFail Text (Either SQL.ConnectionError SomeException) | DBListenFail Text (Either SQL.ConnectionError SomeException)
| DBListenRetry Int | DBListenRetry Int
| DBListenBugHint -- https://github.com/PostgREST/postgrest/issues/3147 | DBListenBugCallQueryFix
| DBListenerGotSCacheMsg ByteString | DBListenerGotSCacheMsg ByteString
| DBListenerGotConfigMsg ByteString | DBListenerGotConfigMsg ByteString
| DBListenerConnectionCleanupFail SomeException | DBListenerConnectionCleanupFail SomeException
@@ -122,8 +121,8 @@ observationMessage = \case
either showListenerConnError showListenerException listenErr either showListenerConnError showListenerException listenErr
DBListenRetry delay -> DBListenRetry delay ->
"Retrying listening for database notifications in " <> (show delay::Text) <> " seconds..." "Retrying listening for database notifications in " <> (show delay::Text) <> " seconds..."
DBListenBugHint -> DBListenBugCallQueryFix ->
"HINT: This is likely a bug in the notification queue, try executing the following to solve it: select pg_notification_queue_usage();" "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 -> DBListenerGotSCacheMsg channel ->
"Received a schema cache reload message on the " <> show channel <> " channel" "Received a schema cache reload message on the " <> show channel <> " channel"
DBListenerGotConfigMsg channel -> DBListenerGotConfigMsg channel ->
@@ -188,6 +187,3 @@ observationMessage = \case
showOnSingleLine :: Char -> Text -> Text 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" 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