refactor: Listener to own module

docs/explanations/architecture.rst

@@ -61,3 +61,14 @@ Admin
 -----
 
 `Admin.hs <https://github.com/PostgREST/postgrest/blob/main/src/PostgREST/Admin.hs>`_ is in charge of the :ref:`admin_server`.
+
+HTTP
+----
+
+The HTTP server is provided by `Warp <https://aosabook.org/en/posa/warp.html>`_.
+
+Listener
+--------
+
+`Listener.hs <https://github.com/PostgREST/postgrest/blob/main/src/PostgREST/Listener.hs>`_ is in charge of maintaining a `LISTEN session <https://www.postgresql.org/docs/current/sql-listen.html>`_
+that keeps the :ref:`schema_cache` and the :ref:`in_db_config` up to date.

postgrest.cabal

@@ -62,6 +62,7 @@ library
                       PostgREST.SchemaCache.Representations
                       PostgREST.SchemaCache.Table
                       PostgREST.Error
+                      PostgREST.Listener
                       PostgREST.Logger
                       PostgREST.MediaType
                       PostgREST.Metrics

src/PostgREST/App.hs

@@ -33,6 +33,7 @@ import qualified PostgREST.AppState   as AppState
 import qualified PostgREST.Auth       as Auth
 import qualified PostgREST.Cors       as Cors
 import qualified PostgREST.Error      as Error
+import qualified PostgREST.Listener   as Listener
 import qualified PostgREST.Logger     as Logger
 import qualified PostgREST.Plan       as Plan
 import qualified PostgREST.Query      as Query
@@ -67,10 +68,10 @@ run appState = do
 
   observer $ AppStartObs prettyVersion
 
-  AppState.connectionWorker appState -- Loads the initial SchemaCache
+  AppState.connectionWorker appState
   Unix.installSignalHandlers (AppState.getMainThreadId appState) (AppState.connectionWorker appState) (AppState.reReadConfig False appState)
-  -- reload schema cache + config on NOTIFY
+
-  AppState.runListener appState
+  Listener.runListener appState
 
   Admin.runAdmin appState (serverSettings conf)
 

src/PostgREST/AppState.hs

@@ -1,5 +1,4 @@
 {-# LANGUAGE LambdaCase      #-}
-{-# LANGUAGE MultiWayIf      #-}
 {-# LANGUAGE NamedFieldPuns  #-}
 {-# LANGUAGE RecordWildCards #-}
 
@@ -21,10 +20,10 @@ module PostgREST.AppState
   , initWithPool
   , putSchemaCache
   , putPgVersion
+  , putIsListenerOn
   , usePool
   , reReadConfig
   , connectionWorker
-  , runListener
   , getObserver
   , isLoaded
   , isPending
@@ -36,8 +35,6 @@ import qualified Data.ByteString.Char8      as BS
 import qualified Data.Cache                 as C
 import           Data.Either.Combinators    (whenLeft)
 import qualified Data.Text                  as T (unpack)
-import           Hasql.Connection           (acquire)
-import qualified Hasql.Notifications        as SQL
 import qualified Hasql.Pool                 as SQL
 import qualified Hasql.Pool.Config          as SQL
 import qualified Hasql.Session              as SQL
@@ -54,9 +51,8 @@ import           System.TimeIt              (timeItT)
 import Control.AutoUpdate (defaultUpdateSettings, mkAutoUpdate,
                            updateAction)
 import Control.Debounce
-import Control.Exception  (throw)
 import Control.Retry      (RetryPolicy, RetryStatus (..), capDelay,
-                           exponentialBackoff, recoverAll, retrying,
+                           exponentialBackoff, retrying,
                            rsPreviousDelay)
 import Data.IORef         (IORef, atomicWriteIORef, newIORef,
                            readIORef)
@@ -64,7 +60,6 @@ import Data.Time.Clock    (UTCTime, getCurrentTime)
 
 import PostgREST.Config                  (AppConfig (..),
                                           addFallbackAppName,
-                                          addTargetSessionAttrs,
                                           readAppConfig)
 import PostgREST.Config.Database         (queryDbSettings,
                                           queryPgVersion,
@@ -442,15 +437,6 @@ 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.
 --
@@ -459,6 +445,8 @@ retryPolicy = capDelay delayMicroseconds $ exponentialBackoff oneSecondInUs
 -- Which might not happen if the server is busy with requests. No idle
 -- connection, no pool timeout.
 --
+-- It's also necessary to release the pool connections because they cache the pg catalog(see #2620)
+--
 -- The connection tries are capped, but if the connection times out no error is
 -- thrown, just 'False' is returned.
 establishConnection :: AppState -> IO ConnectionStatus
@@ -489,6 +477,14 @@ establishConnection appState@AppState{stateObserver=observer} =
       when itShould $ putRetryNextIn appState delay
       return itShould
 
+    retryPolicy :: RetryPolicy
+    retryPolicy =
+      let
+        delayMicroseconds = 32000000 -- 32 seconds
+      in
+      capDelay delayMicroseconds $ exponentialBackoff oneSecondInUs
+    oneSecondInUs = 1000000 -- | One second in microseconds
+
 -- | Re-reads the config plus config options from the db
 reReadConfig :: Bool -> AppState -> IO ()
 reReadConfig startingUp appState@AppState{stateObserver=observer} = do
@@ -526,70 +522,3 @@ reReadConfig startingUp appState@AppState{stateObserver=observer} = do
         pass
       else
         observer ConfigSucceededObs
-
--- | Starts the Listener in a thread
-runListener :: AppState -> IO ()
-runListener appState = do
-  AppConfig{..} <- getConfig appState
-  when configDbChannelEnabled $
-    void . forkIO $ retryingListen appState
-
--- | Starts a LISTEN connection and handles notifications. It recovers with exponential backoff if the LISTEN connection is lost.
--- 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.
--- This is because `Hasql.Notifications.waitForNotifications` uses a forever loop that only finishes when it throws an exception.
-retryingListen :: AppState -> IO ()
-retryingListen appState@AppState{stateObserver=observer, stateMainThreadId=mainThreadId} = do
-  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)
-
-  recoverAll retryPolicy (\RetryStatus{rsIterNumber, rsPreviousDelay} -> do
-
-    when (rsIterNumber > 0) $
-      let delay = fromMaybe 0 rsPreviousDelay `div` oneSecondInUs in
-      observer $ DBListenRetry delay
-
-    connection <- acquire $ toUtf8 (addTargetSessionAttrs $ addFallbackAppName prettyVersion configDbUri)
-    case connection of
-      Right conn -> do
-
-        tryRethrow $ SQL.listen conn $ SQL.toPgIdentifier dbChannel
-
-        putIsListenerOn appState True
-        observer $ DBListenStart dbChannel
-
-        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
-    handleNotification channel msg =
-      if | BS.null msg            -> observer (DBListenerGotSCacheMsg channel) >> cacheReloader
-         | msg == "reload schema" -> observer (DBListenerGotSCacheMsg channel) >> cacheReloader
-         | msg == "reload config" -> observer (DBListenerGotConfigMsg channel) >> reReadConfig False appState
-         | otherwise              -> pure () -- Do nothing if anything else than an empty message is sent
-
-    cacheReloader =
-      -- 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/Auth.hs

@@ -1,9 +1,9 @@
 {-|
 Module      : PostgREST.Auth
-Description : PostgREST authorization functions.
+Description : PostgREST authentication functions.
 
-This module provides functions to deal with the JWT authorization (http://jwt.io).
+This module provides functions to deal with the JWT authentication (http://jwt.io).
-It also can be used to define other authorization functions,
+It also can be used to define other authentication functions,
 in the future Oauth, LDAP and similar integrations can be coded here.
 
 Authentication should always be implemented in an external service.

src/PostgREST/Listener.hs

@@ -0,0 +1,101 @@
+{-# LANGUAGE MultiWayIf      #-}
+{-# LANGUAGE NamedFieldPuns  #-}
+{-# LANGUAGE RecordWildCards #-}
+
+module PostgREST.Listener (runListener) where
+
+import qualified Data.ByteString.Char8 as BS
+
+import Control.Exception       (throw)
+import Data.Either.Combinators (whenLeft)
+
+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 (..))
+import           PostgREST.Version     (prettyVersion)
+
+import           Control.Retry      (RetryPolicy, RetryStatus (..),
+                                     capDelay, exponentialBackoff,
+                                     recoverAll, rsPreviousDelay)
+import qualified PostgREST.AppState as AppState
+import qualified PostgREST.Config   as Config
+
+import Protolude
+
+-- | Starts the Listener in a thread
+runListener :: AppState -> IO ()
+runListener appState = do
+  AppConfig{..} <- getConfig appState
+  when configDbChannelEnabled $
+    void . forkIO $ retryingListen appState
+
+-- | Starts a LISTEN connection and handles notifications. It recovers with exponential backoff if the LISTEN connection is lost.
+-- 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.
+-- This is because `Hasql.Notifications.waitForNotifications` uses a forever loop that only finishes when it throws an exception.
+retryingListen :: AppState -> IO ()
+retryingListen appState = do
+  AppConfig{..} <- AppState.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
+        AppState.putIsListenerOn appState False
+        observer $ DBListenFail dbChannel (Right $ Left ex)
+        unless configDbPoolAutomaticRecovery $ do
+          killThread mainThreadId
+        throw ex)
+
+  recoverAll retryPolicy (\RetryStatus{rsIterNumber, rsPreviousDelay} -> do
+
+    when (rsIterNumber > 0) $
+      let delay = fromMaybe 0 rsPreviousDelay `div` oneSecondInUs in
+      observer $ DBListenRetry delay
+
+    connection <- SQL.acquire $ toUtf8 (Config.addTargetSessionAttrs $ Config.addFallbackAppName prettyVersion configDbUri)
+    case connection of
+      Right conn -> do
+
+        tryRethrow $ SQL.listen conn $ SQL.toPgIdentifier dbChannel
+
+        AppState.putIsListenerOn appState True
+        observer $ DBListenStart dbChannel
+
+        when (rsIterNumber > 0) $ do
+          -- once we can LISTEN again, we might have lost schema cache notificacions, so reload
+          AppState.connectionWorker appState
+
+        tryRethrow $ SQL.waitForNotifications handleNotification conn
+
+      Left err -> do
+        observer $ DBListenFail dbChannel (Left err)
+        -- throw an exception so recoverAll works
+        exitFailure
+    )
+
+  where
+    handleNotification channel msg =
+      if | BS.null msg            -> observer (DBListenerGotSCacheMsg channel) >> cacheReloader
+         | msg == "reload schema" -> observer (DBListenerGotSCacheMsg channel) >> cacheReloader
+         | msg == "reload config" -> observer (DBListenerGotConfigMsg channel) >> AppState.reReadConfig False appState
+         | otherwise              -> pure () -- Do nothing if anything else than an empty message is sent
+
+    cacheReloader =
+      AppState.connectionWorker appState
+
+    observer = AppState.getObserver appState
+    mainThreadId = AppState.getMainThreadId appState
+
+    retryPolicy :: RetryPolicy
+    retryPolicy =
+      let
+        delayMicroseconds = 32000000 -- 32 seconds
+      in
+      capDelay delayMicroseconds $ exponentialBackoff oneSecondInUs
+    oneSecondInUs = 1000000 -- | One second in microseconds