jimmy/postgrest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: make nextListenerDelay local to Listener module

Browse Source 
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.
This commit is contained in:
Michał Kłeczek
2026-06-14 18:39:59 +00:00
committed by Wolfgang Walther
parent 2976eb047b
commit fae6253932
2 changed files with 26 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/PostgREST/AppState.hs
-11
View File
@@ -11,13 +11,11 @@ module PostgREST.AppState
, getMainThreadId , getMainThreadId
, getPgVersion , getPgVersion
, getNextDelay , getNextDelay
, getNextListenerDelay
, getTime , getTime
, getJwtCacheState , getJwtCacheState
, init , init
, initWithPool , initWithPool
, putConfig -- For tests TODO refactoring , putConfig -- For tests TODO refactoring
, putNextListenerDelay
, putSchemaCache , putSchemaCache
, putPgVersion , putPgVersion
, putIsListenerOn , putIsListenerOn
@@ -91,8 +89,6 @@ data AppState = AppState
, stateMainThreadId :: ThreadId , stateMainThreadId :: ThreadId
-- | Keeps track of the next delay for db connection retry -- | Keeps track of the next delay for db connection retry
, stateNextDelay :: IORef Int , stateNextDelay :: IORef Int
-- | Keeps track of the next delay for the listener
, stateNextListenerDelay :: IORef Int
-- | Observation handler -- | Observation handler
, stateObserver :: ObservationHandler , stateObserver :: ObservationHandler
-- | JWT Cache -- | JWT Cache
@@ -131,7 +127,6 @@ initWithPool pool conf loggerState metricsState observer = mdo
<*> mkAutoUpdate defaultUpdateSettings { updateAction = getCurrentTime } <*> mkAutoUpdate defaultUpdateSettings { updateAction = getCurrentTime }
<*> myThreadId <*> myThreadId
<*> newIORef 0 <*> newIORef 0
<*> newIORef 1
<*> pure observer <*> pure observer
<*> JwtCache.init conf observer <*> JwtCache.init conf observer
<*> pure loggerState <*> pure loggerState
@@ -248,12 +243,6 @@ schemaCacheLoader = debouncedSCacheLoader
getNextDelay :: AppState -> IO Int getNextDelay :: AppState -> IO Int
getNextDelay = readIORef . stateNextDelay getNextDelay = readIORef . stateNextDelay
getNextListenerDelay :: AppState -> IO Int
getNextListenerDelay = readIORef . stateNextListenerDelay
putNextListenerDelay :: AppState -> Int -> IO ()
putNextListenerDelay = atomicWriteIORef . stateNextListenerDelay
getConfig :: AppState -> IO AppConfig getConfig :: AppState -> IO AppConfig
getConfig = readIORef . stateConf getConfig = readIORef . stateConf
src/PostgREST/Listener.hs
+12 -9
View File
@@ -18,6 +18,8 @@ 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 Data.IORef (IORef, newIORef,
readIORef, writeIORef)
import qualified Data.Text as T 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
@@ -29,13 +31,14 @@ import Protolude
runListener :: AppState -> IO () runListener :: AppState -> IO ()
runListener appState = do runListener appState = do
AppConfig{..} <- getConfig appState AppConfig{..} <- getConfig appState
when configDbChannelEnabled $ when configDbChannelEnabled $ do
void . forkIO . void $ retryingListen appState False 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. -- | 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 -> Bool -> IO Void retryingListen :: AppState -> IORef Int -> Bool -> IO Void
retryingListen appState hasDbListenerBug = do retryingListen appState nextDelay hasDbListenerBug = do
cfg@AppConfig{..} <- AppState.getConfig appState cfg@AppConfig{..} <- AppState.getConfig appState
let let
dbChannel = toS configDbChannel dbChannel = toS configDbChannel
@@ -48,13 +51,13 @@ retryingListen appState hasDbListenerBug = do
killThread mainThreadId killThread mainThreadId
-- retry the listener -- retry the listener
delay <- AppState.getNextListenerDelay appState delay <- readIORef nextDelay
observer $ DBListenRetry delay observer $ DBListenRetry delay
threadDelay (delay * oneSecondInMicro) threadDelay (delay * oneSecondInMicro)
unless (delay == maxDelay) $ unless (delay == maxDelay) $
AppState.putNextListenerDelay appState (delay * 2) writeIORef nextDelay (delay * 2)
-- loop running the listener -- loop running the listener
retryingListen appState (isDbListenerBug err) retryingListen appState nextDelay (isDbListenerBug err)
-- Execute the listener with error handling -- Execute the listener with error handling
handle onError $ do handle onError $ do
@@ -75,12 +78,12 @@ retryingListen appState hasDbListenerBug = do
AppState.putIsListenerOn appState True AppState.putIsListenerOn appState True
delay <- AppState.getNextListenerDelay appState delay <- readIORef nextDelay
when (delay > 1) $ do -- if we did a retry when (delay > 1) $ do -- if we did a retry
-- assume we lost notifications, refresh the schema cache -- assume we lost notifications, refresh the schema cache
AppState.schemaCacheLoader appState AppState.schemaCacheLoader appState
-- reset the delay -- reset the delay
AppState.putNextListenerDelay appState 1 writeIORef nextDelay 1
observer $ DBListenStart pqHost pqPort pgFullName dbChannel observer $ DBListenStart pqHost pqPort pgFullName dbChannel