From 47e9a2d1341e49e0fb945a8081ebdd39511c3601 Mon Sep 17 00:00:00 2001 From: steve-chavez Date: Thu, 23 May 2024 17:59:20 -0500 Subject: [PATCH] refactor: Listener to own module --- docs/explanations/architecture.rst | 11 ++++ postgrest.cabal | 1 + src/PostgREST/App.hs | 7 +- src/PostgREST/AppState.hs | 95 ++++----------------------- src/PostgREST/Auth.hs | 6 +- src/PostgREST/Listener.hs | 101 +++++++++++++++++++++++++++++ 6 files changed, 132 insertions(+), 89 deletions(-) create mode 100644 src/PostgREST/Listener.hs diff --git a/docs/explanations/architecture.rst b/docs/explanations/architecture.rst index e926f2c19..71af6a0bb 100644 --- a/docs/explanations/architecture.rst +++ b/docs/explanations/architecture.rst @@ -61,3 +61,14 @@ Admin ----- `Admin.hs `_ is in charge of the :ref:`admin_server`. + +HTTP +---- + +The HTTP server is provided by `Warp `_. + +Listener +-------- + +`Listener.hs `_ is in charge of maintaining a `LISTEN session `_ +that keeps the :ref:`schema_cache` and the :ref:`in_db_config` up to date. diff --git a/postgrest.cabal b/postgrest.cabal index fd7744f29..b98c0272a 100644 --- a/postgrest.cabal +++ b/postgrest.cabal @@ -62,6 +62,7 @@ library PostgREST.SchemaCache.Representations PostgREST.SchemaCache.Table PostgREST.Error + PostgREST.Listener PostgREST.Logger PostgREST.MediaType PostgREST.Metrics diff --git a/src/PostgREST/App.hs b/src/PostgREST/App.hs index ca681f50a..d7a3d4eac 100644 --- a/src/PostgREST/App.hs +++ b/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) diff --git a/src/PostgREST/AppState.hs b/src/PostgREST/AppState.hs index a9c5f1947..62d63f8e9 100644 --- a/src/PostgREST/AppState.hs +++ b/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 diff --git a/src/PostgREST/Auth.hs b/src/PostgREST/Auth.hs index ec3095079..37e4fad67 100644 --- a/src/PostgREST/Auth.hs +++ b/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). -It also can be used to define other authorization functions, +This module provides functions to deal with the JWT authentication (http://jwt.io). +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. diff --git a/src/PostgREST/Listener.hs b/src/PostgREST/Listener.hs new file mode 100644 index 000000000..9a8f68fe0 --- /dev/null +++ b/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