fix: retry connection on failed schema cache load (#1685)
Retry the connection when the "Failed to load the schema cache" error happens. Also die if the schema cache query has a syntax error.
This commit is contained in:
+28
-11
@@ -35,7 +35,8 @@ import PostgREST.DbStructure (getDbStructure, getPgVersion)
|
|||||||
import PostgREST.Error (PgError (PgError), checkIsFatal,
|
import PostgREST.Error (PgError (PgError), checkIsFatal,
|
||||||
errorPayload)
|
errorPayload)
|
||||||
import PostgREST.Types (ConnectionStatus (..), DbStructure,
|
import PostgREST.Types (ConnectionStatus (..), DbStructure,
|
||||||
PgVersion (..), minimumPgVersion)
|
PgVersion (..), SCacheStatus (..),
|
||||||
|
minimumPgVersion)
|
||||||
import Protolude hiding (hPutStrLn, head, toS)
|
import Protolude hiding (hPutStrLn, head, toS)
|
||||||
import Protolude.Conv (toS)
|
import Protolude.Conv (toS)
|
||||||
|
|
||||||
@@ -177,7 +178,7 @@ _1s = 1000000 :: Int -- 1 second
|
|||||||
1. Tries to connect to pg server and will keep trying until success.
|
1. Tries to connect to pg server and will keep trying until success.
|
||||||
2. Checks if the pg version is supported and if it's not it kills the main
|
2. Checks if the pg version is supported and if it's not it kills the main
|
||||||
program.
|
program.
|
||||||
3. Obtains the dbStructure.
|
3. Obtains the dbStructure. If this fails, it goes back to 1.
|
||||||
-}
|
-}
|
||||||
connectionWorker
|
connectionWorker
|
||||||
:: ThreadId -- ^ Main thread id. Killed if pg version is unsupported
|
:: ThreadId -- ^ Main thread id. Killed if pg version is unsupported
|
||||||
@@ -203,7 +204,11 @@ connectionWorker mainTid pool refConf refDbStructure refIsWorkerOn (dbChannelEna
|
|||||||
NotConnected -> return () -- Unreachable because connectionStatus will keep trying to connect
|
NotConnected -> return () -- Unreachable because connectionStatus will keep trying to connect
|
||||||
Connected actualPgVersion -> do -- Procede with initialization
|
Connected actualPgVersion -> do -- Procede with initialization
|
||||||
putStrLn ("Connection successful" :: Text)
|
putStrLn ("Connection successful" :: Text)
|
||||||
fillSchemaCache pool actualPgVersion refConf refDbStructure
|
scStatus <- loadSchemaCache pool actualPgVersion refConf refDbStructure
|
||||||
|
case scStatus of
|
||||||
|
SCLoaded -> pure () -- do nothing and proceed if the load was successful
|
||||||
|
SCOnRetry -> work -- retry
|
||||||
|
SCFatalFail -> killThread mainTid -- die if our schema cache query has an error
|
||||||
liftIO $ atomicWriteIORef refIsWorkerOn False
|
liftIO $ atomicWriteIORef refIsWorkerOn False
|
||||||
|
|
||||||
{-|
|
{-|
|
||||||
@@ -244,21 +249,29 @@ connectionStatus pool =
|
|||||||
putStrLn $ "Attempting to reconnect to the database in " <> (show delay::Text) <> " seconds..."
|
putStrLn $ "Attempting to reconnect to the database in " <> (show delay::Text) <> " seconds..."
|
||||||
return itShould
|
return itShould
|
||||||
|
|
||||||
-- | Fill the DbStructure by using a connection from the pool
|
-- | Load the DbStructure by using a connection from the pool.
|
||||||
fillSchemaCache :: P.Pool -> PgVersion -> IORef AppConfig -> IORef (Maybe DbStructure) -> IO ()
|
loadSchemaCache :: P.Pool -> PgVersion -> IORef AppConfig -> IORef (Maybe DbStructure) -> IO SCacheStatus
|
||||||
fillSchemaCache pool actualPgVersion refConf refDbStructure = do
|
loadSchemaCache pool actualPgVersion refConf refDbStructure = do
|
||||||
conf <- readIORef refConf
|
conf <- readIORef refConf
|
||||||
result <- P.use pool $ HT.transaction HT.ReadCommitted HT.Read $ getDbStructure (toList $ configDbSchemas conf) (configDbExtraSearchPath conf) actualPgVersion (configDbPreparedStatements conf)
|
result <- P.use pool $ HT.transaction HT.ReadCommitted HT.Read $ getDbStructure (toList $ configDbSchemas conf) (configDbExtraSearchPath conf) actualPgVersion (configDbPreparedStatements conf)
|
||||||
case result of
|
case result of
|
||||||
Left e -> do
|
Left e -> do
|
||||||
-- If this error happens it would mean the connection is down again. Improbable because connectionStatus ensured the connection.
|
let err = PgError False e
|
||||||
-- It's not a problem though, because App.postgrest would retry the connectionWorker or the user can do a SIGSUR1 again.
|
putErr = hPutStrLn stderr . toS . errorPayload $ err
|
||||||
hPutStrLn stderr . toS . errorPayload $ PgError False e
|
case checkIsFatal err of
|
||||||
putStrLn ("Failed to load the schema cache" :: Text)
|
Just _ -> do
|
||||||
|
hPutStrLn stderr ("A fatal error ocurred when loading the schema cache" :: Text)
|
||||||
|
putErr
|
||||||
|
hPutStrLn stderr ("This is probably a bug in PostgREST, please report it at https://github.com/PostgREST/postgrest/issues" :: Text)
|
||||||
|
return SCFatalFail
|
||||||
|
Nothing -> do
|
||||||
|
hPutStrLn stderr ("An error ocurred when loading the schema cache" :: Text) >> putErr
|
||||||
|
return SCOnRetry
|
||||||
|
|
||||||
Right dbStructure -> do
|
Right dbStructure -> do
|
||||||
atomicWriteIORef refDbStructure $ Just dbStructure
|
atomicWriteIORef refDbStructure $ Just dbStructure
|
||||||
putStrLn ("Schema cache loaded" :: Text)
|
putStrLn ("Schema cache loaded" :: Text)
|
||||||
|
return SCLoaded
|
||||||
|
|
||||||
{-|
|
{-|
|
||||||
Starts a dedicated pg connection to LISTEN for notifications.
|
Starts a dedicated pg connection to LISTEN for notifications.
|
||||||
@@ -274,8 +287,12 @@ listener dbUri dbChannel pool refConf refDbStructure mvarConnectionStatus connWo
|
|||||||
Connected actualPgVersion -> void $ forkFinally (do -- forkFinally allows to detect if the thread dies
|
Connected actualPgVersion -> void $ forkFinally (do -- forkFinally allows to detect if the thread dies
|
||||||
dbOrError <- C.acquire dbUri
|
dbOrError <- C.acquire dbUri
|
||||||
-- Debounce in case too many NOTIFYs arrive. Could happen on a migration(assuming a pg EVENT TRIGGER is set up).
|
-- Debounce in case too many NOTIFYs arrive. Could happen on a migration(assuming a pg EVENT TRIGGER is set up).
|
||||||
|
-- This might not be needed according to pg docs https://www.postgresql.org/docs/12/sql-notify.html:
|
||||||
|
-- "If the same channel name is signaled multiple times from the same transaction with identical payload strings, the database server can decide to deliver a single notification only."
|
||||||
|
-- But we do it to be extra safe.
|
||||||
scFiller <- mkDebounce (defaultDebounceSettings {
|
scFiller <- mkDebounce (defaultDebounceSettings {
|
||||||
debounceAction = fillSchemaCache pool actualPgVersion refConf refDbStructure,
|
-- It's not necessary to check the loadSchemaCache success here. If the connection drops, the thread will die and proceed to recover below.
|
||||||
|
debounceAction = void $ loadSchemaCache pool actualPgVersion refConf refDbStructure,
|
||||||
debounceEdge = trailingEdge, -- wait until the function hasn’t been called in _1s
|
debounceEdge = trailingEdge, -- wait until the function hasn’t been called in _1s
|
||||||
debounceFreq = _1s })
|
debounceFreq = _1s })
|
||||||
case dbOrError of
|
case dbOrError of
|
||||||
|
|||||||
@@ -115,7 +115,7 @@ compressedRel rel =
|
|||||||
(_, _, _) ->
|
(_, _, _) ->
|
||||||
mempty
|
mempty
|
||||||
|
|
||||||
data PgError = PgError Authenticated P.UsageError
|
data PgError = PgError Authenticated P.UsageError deriving Show
|
||||||
type Authenticated = Bool
|
type Authenticated = Bool
|
||||||
|
|
||||||
instance PgrstError PgError where
|
instance PgrstError PgError where
|
||||||
@@ -216,7 +216,9 @@ checkIsFatal (PgError _ (P.ConnectionError e))
|
|||||||
| isAuthFailureMessage = Just $ toS failureMessage
|
| isAuthFailureMessage = Just $ toS failureMessage
|
||||||
| otherwise = Nothing
|
| otherwise = Nothing
|
||||||
where isAuthFailureMessage = "FATAL: password authentication failed" `isPrefixOf` toS failureMessage
|
where isAuthFailureMessage = "FATAL: password authentication failed" `isPrefixOf` toS failureMessage
|
||||||
failureMessage = fromMaybe "" e
|
failureMessage = fromMaybe mempty e
|
||||||
|
-- Chek for a syntax error(42601 is the pg code). This would mean the error is on our part somehow, so we treat it as fatal.
|
||||||
|
checkIsFatal (PgError _ (P.SessionError (H.QueryError _ _ (H.ResultError (H.ServerError "42601" e _ _))))) = Just $ toS e
|
||||||
checkIsFatal _ = Nothing
|
checkIsFatal _ = Nothing
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -560,6 +560,13 @@ data ConnectionStatus
|
|||||||
| FatalConnectionError Text
|
| FatalConnectionError Text
|
||||||
deriving (Eq, Show)
|
deriving (Eq, Show)
|
||||||
|
|
||||||
|
-- | Schema cache status
|
||||||
|
data SCacheStatus
|
||||||
|
= SCLoaded
|
||||||
|
| SCOnRetry
|
||||||
|
| SCFatalFail
|
||||||
|
deriving (Eq, Show)
|
||||||
|
|
||||||
data LogLevel = LogCrit | LogError | LogWarn | LogInfo deriving (Eq)
|
data LogLevel = LogCrit | LogError | LogWarn | LogInfo deriving (Eq)
|
||||||
|
|
||||||
instance Show LogLevel where
|
instance Show LogLevel where
|
||||||
|
|||||||
Reference in New Issue
Block a user