From fae625393241e27115a805c3ecee11cd7554373b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20K=C5=82eczek?= Date: Sun, 14 Jun 2026 08:22:44 +0200 Subject: [PATCH] refactor: make nextListenerDelay local to Listener module Right now stateNextListenerDelay is only used in Listener module. This change is a small refactoring moving nextListenerDelay to Listener to increase cohesion and decrease coupling. --- src/PostgREST/AppState.hs | 39 ++++++++++++++------------------------- src/PostgREST/Listener.hs | 21 ++++++++++++--------- 2 files changed, 26 insertions(+), 34 deletions(-) diff --git a/src/PostgREST/AppState.hs b/src/PostgREST/AppState.hs index 74126ffb5..93dff4134 100644 --- a/src/PostgREST/AppState.hs +++ b/src/PostgREST/AppState.hs @@ -11,13 +11,11 @@ module PostgREST.AppState , getMainThreadId , getPgVersion , getNextDelay - , getNextListenerDelay , getTime , getJwtCacheState , init , initWithPool , putConfig -- For tests TODO refactoring - , putNextListenerDelay , putSchemaCache , putPgVersion , putIsListenerOn @@ -72,33 +70,31 @@ import Protolude data AppState = AppState -- | Database connection pool - { statePool :: SQL.Pool + { statePool :: SQL.Pool -- | Database server version - , statePgVersion :: IORef PgVersion + , statePgVersion :: IORef PgVersion -- | Schema cache - , stateSchemaCache :: IORef (Maybe SchemaCache) + , stateSchemaCache :: IORef (Maybe SchemaCache) -- | The schema cache status - , stateSCacheStatus :: SchemaCacheStatus + , stateSCacheStatus :: SchemaCacheStatus -- | State of the LISTEN channel - , stateIsListenerOn :: IORef Bool + , stateIsListenerOn :: IORef Bool -- | starts the connection worker with a debounce - , debouncedSCacheLoader :: IO () + , debouncedSCacheLoader :: IO () -- | Config that can change at runtime - , stateConf :: IORef AppConfig + , stateConf :: IORef AppConfig -- | Time used for verifying JWT expiration - , stateGetTime :: IO UTCTime + , stateGetTime :: IO UTCTime -- | Used for killing the main thread in case a subthread fails - , stateMainThreadId :: ThreadId + , stateMainThreadId :: ThreadId -- | Keeps track of the next delay for db connection retry - , stateNextDelay :: IORef Int - -- | Keeps track of the next delay for the listener - , stateNextListenerDelay :: IORef Int + , stateNextDelay :: IORef Int -- | Observation handler - , stateObserver :: ObservationHandler + , stateObserver :: ObservationHandler -- | JWT Cache - , stateJwtCache :: JwtCache.JwtCacheState - , stateLogger :: Logger.LoggerState - , stateMetrics :: Metrics.MetricsState + , stateJwtCache :: JwtCache.JwtCacheState + , stateLogger :: Logger.LoggerState + , stateMetrics :: Metrics.MetricsState } -- | Schema cache status. @@ -131,7 +127,6 @@ initWithPool pool conf loggerState metricsState observer = mdo <*> mkAutoUpdate defaultUpdateSettings { updateAction = getCurrentTime } <*> myThreadId <*> newIORef 0 - <*> newIORef 1 <*> pure observer <*> JwtCache.init conf observer <*> pure loggerState @@ -248,12 +243,6 @@ schemaCacheLoader = debouncedSCacheLoader getNextDelay :: AppState -> IO Int getNextDelay = readIORef . stateNextDelay -getNextListenerDelay :: AppState -> IO Int -getNextListenerDelay = readIORef . stateNextListenerDelay - -putNextListenerDelay :: AppState -> Int -> IO () -putNextListenerDelay = atomicWriteIORef . stateNextListenerDelay - getConfig :: AppState -> IO AppConfig getConfig = readIORef . stateConf diff --git a/src/PostgREST/Listener.hs b/src/PostgREST/Listener.hs index 25ba02027..c3a4beeb5 100644 --- a/src/PostgREST/Listener.hs +++ b/src/PostgREST/Listener.hs @@ -18,6 +18,8 @@ import qualified PostgREST.Config as Config import Control.Arrow ((&&&)) import Data.Bitraversable (bisequence) import Data.Either.Combinators (whenRight) +import Data.IORef (IORef, newIORef, + readIORef, writeIORef) import qualified Data.Text as T import qualified Database.PostgreSQL.LibPQ as LibPQ import qualified Hasql.Session as SQL @@ -29,13 +31,14 @@ import Protolude runListener :: AppState -> IO () runListener appState = do AppConfig{..} <- getConfig appState - when configDbChannelEnabled $ - void . forkIO . void $ retryingListen appState False + when configDbChannelEnabled $ do + nextDelay <- newIORef 1 + void . forkIO . void $ retryingListen appState nextDelay 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 -> Bool -> IO Void -retryingListen appState hasDbListenerBug = do +retryingListen :: AppState -> IORef Int -> Bool -> IO Void +retryingListen appState nextDelay hasDbListenerBug = do cfg@AppConfig{..} <- AppState.getConfig appState let dbChannel = toS configDbChannel @@ -48,13 +51,13 @@ retryingListen appState hasDbListenerBug = do killThread mainThreadId -- retry the listener - delay <- AppState.getNextListenerDelay appState + delay <- readIORef nextDelay observer $ DBListenRetry delay threadDelay (delay * oneSecondInMicro) unless (delay == maxDelay) $ - AppState.putNextListenerDelay appState (delay * 2) + writeIORef nextDelay (delay * 2) -- loop running the listener - retryingListen appState (isDbListenerBug err) + retryingListen appState nextDelay (isDbListenerBug err) -- Execute the listener with error handling handle onError $ do @@ -75,12 +78,12 @@ retryingListen appState hasDbListenerBug = do AppState.putIsListenerOn appState True - delay <- AppState.getNextListenerDelay appState + delay <- readIORef nextDelay when (delay > 1) $ do -- if we did a retry -- assume we lost notifications, refresh the schema cache AppState.schemaCacheLoader appState -- reset the delay - AppState.putNextListenerDelay appState 1 + writeIORef nextDelay 1 observer $ DBListenStart pqHost pqPort pgFullName dbChannel