fix: listener retries with exponential backoff

Also corrects the admin ready response which now considers the listener
state.

CHANGELOG.md

@@ -37,7 +37,8 @@ This project adheres to [Semantic Versioning](http://semver.org/).
  - #3424, Admin `/live` and `/ready` now differentiates a failure as 500 status - @steve-chavez
     + 503 status is still given when postgREST is in a recovering state
  - #3478, Media Types are parsed case insensitively - @develop7
- - #2781, Fix listener silently failing on read replica - @steve-chavez
+ - #3533, #3536, Fix listener silently failing on read replica - @steve-chavez
+    + If the LISTEN connection fails, it's retried with exponential backoff
 
 ### Deprecated
 

src/PostgREST/App.hs

@@ -70,7 +70,7 @@ run appState = do
   AppState.connectionWorker appState -- Loads the initial SchemaCache
   Unix.installSignalHandlers (AppState.getMainThreadId appState) (AppState.connectionWorker appState) (AppState.reReadConfig False appState)
   -- reload schema cache + config on NOTIFY
-  AppState.runListener conf appState
+  AppState.runListener appState
 
   Admin.runAdmin appState (serverSettings conf)
 

src/PostgREST/AppState.hs

@@ -54,8 +54,10 @@ import           System.TimeIt              (timeItT)
 import Control.AutoUpdate (defaultUpdateSettings, mkAutoUpdate,
                            updateAction)
 import Control.Debounce
-import Control.Retry      (RetryStatus, capDelay, exponentialBackoff,
+import Control.Exception  (throw)
-                           retrying, rsPreviousDelay)
+import Control.Retry      (RetryPolicy, RetryStatus (..), capDelay,
+                           exponentialBackoff, recoverAll, retrying,
+                           rsPreviousDelay)
 import Data.IORef         (IORef, atomicWriteIORef, newIORef,
                            readIORef)
 import Data.Time.Clock    (UTCTime, getCurrentTime)
@@ -94,10 +96,10 @@ data AppState = AppState
   , stateSCacheStatus         :: IORef SchemaCacheStatus
   -- | The connection status
   , stateConnStatus           :: IORef ConnectionStatus
+  -- | State of the LISTEN channel
+  , stateIsListenerOn         :: IORef Bool
   -- | starts the connection worker with a debounce
   , debouncedConnectionWorker :: IO ()
-  -- | Binary semaphore used to sync the listener(NOTIFY reload) with the connectionWorker.
-  , stateListener             :: MVar ()
   -- | Config that can change at runtime
   , stateConf                 :: IORef AppConfig
   -- | Time used for verifying JWT expiration
@@ -152,8 +154,8 @@ initWithPool (sock, adminSock) pool conf loggerState metricsState observer = do
     <*> newIORef Nothing
     <*> newIORef SCPending
     <*> newIORef ConnPending
+    <*> newIORef False
     <*> pure (pure ())
-    <*> newEmptyMVar
     <*> newIORef conf
     <*> mkAutoUpdate defaultUpdateSettings { updateAction = getCurrentTime }
     <*> myThreadId
@@ -329,16 +331,16 @@ getSocketAdmin = stateSocketAdmin
 getMainThreadId :: AppState -> ThreadId
 getMainThreadId = stateMainThreadId
 
--- | As this IO action uses `takeMVar` internally, it will only return once
+getIsListenerOn :: AppState -> IO Bool
--- `stateListener` has been set using `signalListener`. This is currently used
+getIsListenerOn appState = do
--- to syncronize workers.
+  AppConfig{..} <- getConfig appState
-waitListener :: AppState -> IO ()
+  if configDbChannelEnabled then
-waitListener = takeMVar . stateListener
+    readIORef $ stateIsListenerOn appState
+  else
+    pure True
 
--- tryPutMVar doesn't lock the thread. It should always succeed since
+putIsListenerOn :: AppState -> Bool -> IO ()
--- the connectionWorker is the only mvar producer.
+putIsListenerOn = atomicWriteIORef . stateIsListenerOn
-signalListener :: AppState -> IO ()
-signalListener appState = void $ tryPutMVar (stateListener appState) ()
 
 isConnEstablished :: AppState -> IO Bool
 isConnEstablished x = do
@@ -354,13 +356,15 @@ isLoaded :: AppState -> IO Bool
 isLoaded x = do
   scacheStatus <- readIORef $ stateSCacheStatus x
   connEstablished <- isConnEstablished x
-  return $ scacheStatus == SCLoaded && connEstablished
+  listenerOn <- getIsListenerOn x
+  return $ scacheStatus == SCLoaded && connEstablished && listenerOn
 
 isPending :: AppState -> IO Bool
 isPending x = do
   scacheStatus <- readIORef $ stateSCacheStatus x
   connStatus <- readIORef $ stateConnStatus x
-  return $ scacheStatus == SCPending || connStatus == ConnPending
+  listenerOn <- getIsListenerOn x
+  return $ scacheStatus == SCPending || connStatus == ConnPending || not listenerOn
 
 putSCacheStatus :: AppState -> SchemaCacheStatus -> IO ()
 putSCacheStatus = atomicWriteIORef . stateSCacheStatus
@@ -424,9 +428,6 @@ internalConnectionWorker appState@AppState{stateObserver=observer, stateMainThre
           when (actualPgVersion < minimumPgVersion) $ do
             observer $ ExitUnsupportedPgVersion actualPgVersion minimumPgVersion
             killThread mainThreadId
-          -- Procede with initialization
-          when configDbChannelEnabled $
-            signalListener appState
           observer (DBConnectedObs $ pgvFullName actualPgVersion)
           -- this could be fail because the connection drops, but the loadSchemaCache will pick the error and retry again
           -- We cannot retry after it fails immediately, because db-pre-config could have user errors. We just log the error and continue.
@@ -440,6 +441,15 @@ internalConnectionWorker appState@AppState{stateObserver=observer, stateMainThre
               -- retry reloading the schema cache
               work
 
+-- | One second in microseconds
+oneSecondInUs :: Int
+oneSecondInUs = 1000000
+
+retryPolicy :: RetryPolicy
+retryPolicy = capDelay delayMicroseconds $ exponentialBackoff oneSecondInUs
+  where
+    delayMicroseconds = 32000000 -- 32 seconds
+
 -- | Repeatedly flush the pool, and check if a connection from the
 -- pool allows access to the PostgreSQL database.
 --
@@ -452,13 +462,9 @@ internalConnectionWorker appState@AppState{stateObserver=observer, stateMainThre
 -- thrown, just 'False' is returned.
 establishConnection :: AppState -> IO ConnectionStatus
 establishConnection appState@AppState{stateObserver=observer} =
-  retrying retrySettings shouldRetry $
+  retrying retryPolicy shouldRetry $
     const $ flushPool appState >> getConnectionStatus
   where
-    retrySettings = capDelay delayMicroseconds $ exponentialBackoff backoffMicroseconds
-    delayMicroseconds = 32000000 -- 32 seconds
-    backoffMicroseconds = 1000000 -- 1 second
-
     getConnectionStatus :: IO ConnectionStatus
     getConnectionStatus = do
       pgVersion <- usePool appState (queryPgVersion False) -- No need to prepare the query here, as the connection might not be established
@@ -476,7 +482,7 @@ establishConnection appState@AppState{stateObserver=observer} =
     shouldRetry rs isConnSucc = do
       AppConfig{..} <- getConfig appState
       let
-        delay = fromMaybe 0 (rsPreviousDelay rs) `div` backoffMicroseconds
+        delay = fromMaybe 0 (rsPreviousDelay rs) `div` oneSecondInUs
         itShould = ConnPending == isConnSucc && configDbPoolAutomaticRecovery
       when itShould $ observer $ ConnectionRetryObs delay
       when itShould $ putRetryNextIn appState delay
@@ -520,49 +526,62 @@ reReadConfig startingUp appState@AppState{stateObserver=observer} = do
       else
         observer ConfigSucceededObs
 
-runListener :: AppConfig -> AppState -> IO ()
+-- | Starts the Listener in a thread
-runListener conf@AppConfig{configDbChannelEnabled} appState = do
+runListener :: AppState -> IO ()
-  when configDbChannelEnabled $ listener appState conf
+runListener appState = do
+  AppConfig{..} <- getConfig appState
+  when configDbChannelEnabled $
+    void . forkIO $ retryingListen appState
 
--- | Starts a dedicated pg connection to LISTEN for notifications.  When a
+-- | Starts a LISTEN connection and handles notifications. It recovers with exponential backoff if the LISTEN connection is lost.
--- NOTIFY <db-channel> - with an empty payload - is done, it refills the schema
+-- TODO Once the listen channel is recovered, the retry status is not reset. So if the last backoff was 4 seconds, the next time recovery kicks in the backoff will be 8 seconds.
--- cache.  It uses the connectionWorker in case the LISTEN connection dies.
+-- This is because `Hasql.Notifications.waitForNotifications` uses a forever loop that only finishes when it throws an exception.
-listener :: AppState -> AppConfig -> IO ()
+retryingListen :: AppState -> IO ()
-listener appState@AppState{stateObserver=observer, stateMainThreadId=mainThreadId} conf@AppConfig{..} = do
+retryingListen appState@AppState{stateObserver=observer, stateMainThreadId=mainThreadId} = do
-  let dbChannel = toS configDbChannel
+  AppConfig{..} <- getConfig appState
+  let
+    dbChannel = toS configDbChannel
+    -- Try, catch and rethrow the exception. This is done so we can observe the failure message and let Control.Retry.recoverAll do its work.
+    -- There's a `Control.Retry.recovering` we could use to avoid this rethrowing, but it's more complex to use.
+    -- The root cause of these workarounds is that `Hasql.Notifications.waitForNotifications` uses exceptions.
+    tryRethrow :: IO () -> IO ()
+    tryRethrow action =  do
+      act <- try action
+      whenLeft act (\ex -> do
+        putIsListenerOn appState False
+        observer $ DBListenFail dbChannel (Right $ Left ex)
+        unless configDbPoolAutomaticRecovery $ do
+          killThread mainThreadId
+        throw ex)
 
-  -- The listener has to wait for a signal from the connectionWorker.
+  recoverAll retryPolicy (\RetryStatus{rsIterNumber, rsPreviousDelay} -> do
-  -- This is because when the connection to the db is lost, the listener also
-  -- tries to recover the connection, but not with the same pace as the connectionWorker.
-  -- Not waiting makes stderr quickly fill with connection retries messages from the listener.
-  waitListener appState
 
-  -- forkFinally allows to detect if the thread dies
+    when (rsIterNumber > 0) $
-  void . flip forkFinally (handleFinally dbChannel configDbPoolAutomaticRecovery) $ do
+      let delay = fromMaybe 0 rsPreviousDelay `div` oneSecondInUs in
-    dbOrError <- acquire $ toUtf8 (addFallbackAppName prettyVersion configDbUri)
+      observer $ DBListenRetry delay
-    case dbOrError of
+
-      Right db -> do
+    connection <- acquire $ toUtf8 (addFallbackAppName prettyVersion configDbUri)
-        SQL.listen db $ SQL.toPgIdentifier dbChannel
+    case connection of
+      Right conn -> do
+
+        tryRethrow $ SQL.listen conn $ SQL.toPgIdentifier dbChannel
+
+        putIsListenerOn appState True
         observer $ DBListenStart dbChannel
-        SQL.waitForNotifications handleNotification db
+
+        when (rsIterNumber > 0) $ do
+          -- once we can LISTEN again, we might have lost schema cache notificacions, so reload
+          connectionWorker appState
+
+        tryRethrow $ SQL.waitForNotifications handleNotification conn
 
       Left err -> do
         observer $ DBListenFail dbChannel (Left err)
+        -- throw an exception so recoverAll works
         exitFailure
+    )
+
   where
-    handleFinally dbChannel False err = do
-      observer $ DBListenFail dbChannel (Right err)
-      killThread mainThreadId
-    handleFinally dbChannel True err = do
-      -- if the thread dies, we try to recover
-      observer $ DBListenFail dbChannel (Right err)
-      -- assume the pool connection was also lost, call the connection worker
-      connectionWorker appState
-
-      -- retry the listener
-      observer DBListenRetry
-      listener appState conf
-
     handleNotification channel msg =
       if | BS.null msg            -> observer (DBListenerGotSCacheMsg channel) >> cacheReloader
          | msg == "reload schema" -> observer (DBListenerGotSCacheMsg channel) >> cacheReloader
@@ -573,4 +592,3 @@ listener appState@AppState{stateObserver=observer, stateMainThreadId=mainThreadI
       -- reloads the schema cache + restarts pool connections
       -- it's necessary to restart the pg connections because they cache the pg catalog(see #2620)
       connectionWorker appState
-

src/PostgREST/Observation.hs

@@ -42,7 +42,7 @@ data Observation
   | ConnectionPgVersionErrorObs SQL.UsageError
   | DBListenStart Text
   | DBListenFail Text (Either SQL.ConnectionError (Either SomeException ()))
-  | DBListenRetry
+  | DBListenRetry Int
   | DBListenerGotSCacheMsg ByteString
   | DBListenerGotConfigMsg ByteString
   | ConfigReadErrorObs SQL.UsageError
@@ -105,8 +105,8 @@ observationMessage = \case
         Left err  -> show err
         Right err -> showListenerError err
     )
-  DBListenRetry ->
+  DBListenRetry delay ->
-    "Retrying listening for notifications..."
+    "Retrying listening for notifications in " <> (show delay::Text) <> " seconds..."
   DBListenerGotSCacheMsg channel ->
     "Received a schema cache reload message on the " <> show channel <> " channel"
   DBListenerGotConfigMsg channel ->

test/io/test_io.py

@@ -733,8 +733,8 @@ def test_admin_ready_includes_schema_cache_state(defaultenv, metapostgrest):
 
         # force a reconnection so the new role setting is picked up
         postgrest.process.send_signal(signal.SIGUSR1)
-        # wait 600ms to finish schema cache reload attempt
+
-        time.sleep(0.6)
+        postgrest.wait_until_scache_starts_loading()
 
         response = postgrest.admin.get("/ready", timeout=1)
         assert response.status_code == 503

test/io/test_replica.py

@@ -20,7 +20,12 @@ def test_sanity_replica(replicaenv):
         response = postgrest.session.get("/items?select=count")
         assert response.text == '[{"count":10}]'
 
-    with run(env=replicaenv["replica"]) as postgrest:
+    working_replica_env = {
+        **replicaenv["replica"],
+        "PGRST_DB_CHANNEL_ENABLED": "false",  # LISTEN doesn't work on read replicas
+    }
+
+    with run(env=working_replica_env) as postgrest:
         response = postgrest.session.get("/rpc/is_replica")
         assert response.text == "true"