diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 28afe1cbe..e0933c8dd 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -55,12 +55,6 @@ It builds the OpenAPI response using the schema cache. This module provides functions to deal with JWT authorization. -### Workers.hs - -This spawns threads which are used to execute concurrent jobs. - -Jobs include connection recovery, a listener for the PostgreSQL LISTEN command, and an admin server. - ### SchemaCache.hs This queries the PostgreSQL system catalogs and caches the metadata into a SchemaCache type, @@ -68,3 +62,7 @@ This queries the PostgreSQL system catalogs and caches the metadata into a Schem ### AppState.hs The state of the App which is kept across requests. + +This spawns threads which are used to execute concurrent jobs. + +Jobs include connection recover and a listener for the PostgreSQL LISTEN command. diff --git a/CHANGELOG.md b/CHANGELOG.md index e8bfa6bc6..4110f7f4d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,10 @@ This project adheres to [Semantic Versioning](http://semver.org/). + New config option `db-pre-config`(empty by default) + Allows using the in-database configuration without SUPERUSER +### Fixed + + - #2791, Fix dropping schema cache reload notifications - @steve-chavez + ## [11.0.1] - 2023-04-27 ### Fixed diff --git a/postgrest.cabal b/postgrest.cabal index 743716557..daf8b6c1f 100644 --- a/postgrest.cabal +++ b/postgrest.cabal @@ -34,7 +34,8 @@ library default-extensions: OverloadedStrings NoImplicitPrelude hs-source-dirs: src - exposed-modules: PostgREST.App + exposed-modules: PostgREST.Admin + PostgREST.App PostgREST.AppState PostgREST.Auth PostgREST.CLI @@ -70,7 +71,6 @@ library PostgREST.Response.OpenAPI PostgREST.Response.GucHeader PostgREST.Version - PostgREST.Workers other-modules: Paths_postgrest build-depends: base >= 4.9 && < 4.17 , HTTP >= 4000.3.7 && < 4000.5 @@ -86,6 +86,7 @@ library , contravariant-extras >= 0.3.3 && < 0.4 , cookie >= 0.4.2 && < 0.5 , either >= 4.4.1 && < 5.1 + , extra >= 1.7.0 && < 2.0 , fuzzyset >= 0.2.3 , gitrev >= 1.2 && < 1.4 , hasql >= 1.6.1.1 && < 1.7 diff --git a/src/PostgREST/Admin.hs b/src/PostgREST/Admin.hs new file mode 100644 index 000000000..ffc61cbfb --- /dev/null +++ b/src/PostgREST/Admin.hs @@ -0,0 +1,87 @@ +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE RecordWildCards #-} + +module PostgREST.Admin + ( runAdmin + ) where + +import qualified Data.Text as T +import qualified Hasql.Session as SQL +import qualified Network.HTTP.Types.Status as HTTP +import qualified Network.Wai as Wai +import qualified Network.Wai.Handler.Warp as Warp + +import Control.Monad.Extra (whenJust) + +import Network.Socket +import Network.Socket.ByteString + +import PostgREST.AppState (AppState) +import PostgREST.Config (AppConfig (..)) + +import qualified PostgREST.AppState as AppState + +import Protolude + +runAdmin :: AppConfig -> AppState -> Warp.Settings -> IO () +runAdmin conf@AppConfig{configAdminServerPort} appState settings = + whenJust configAdminServerPort $ \adminPort -> do + AppState.logWithZTime appState $ "Admin server listening on port " <> show adminPort + void . forkIO $ Warp.runSettings (settings & Warp.setPort adminPort) adminApp + where + adminApp = admin appState conf + +-- | PostgREST admin application +admin :: AppState.AppState -> AppConfig -> Wai.Application +admin appState appConfig req respond = do + isMainAppReachable <- any isRight <$> reachMainApp appConfig + isSchemaCacheLoaded <- isJust <$> AppState.getSchemaCache appState + isConnectionUp <- + if configDbChannelEnabled appConfig + then AppState.getIsListenerOn appState + else isRight <$> AppState.usePool appState (SQL.sql "SELECT 1") + + case Wai.pathInfo req of + ["ready"] -> + respond $ Wai.responseLBS (if isMainAppReachable && isConnectionUp && isSchemaCacheLoaded then HTTP.status200 else HTTP.status503) [] mempty + ["live"] -> + respond $ Wai.responseLBS (if isMainAppReachable then HTTP.status200 else HTTP.status503) [] mempty + _ -> + respond $ Wai.responseLBS HTTP.status404 [] mempty + +-- Try to connect to the main app socket +-- Note that it doesn't even send a valid HTTP request, we just want to check that the main app is accepting connections +-- The code for resolving the "*4", "!4", "*6", "!6", "*" special values is taken from +-- https://hackage.haskell.org/package/streaming-commons-0.2.2.4/docs/src/Data.Streaming.Network.html#bindPortGenEx +reachMainApp :: AppConfig -> IO [Either IOException ()] +reachMainApp AppConfig{..} = + case configServerUnixSocket of + Just path -> do + sock <- socket AF_UNIX Stream 0 + (:[]) <$> try (do + connect sock $ SockAddrUnix path + withSocketsDo $ bracket (pure sock) close sendEmpty) + Nothing -> do + let + host | configServerHost `elem` ["*4", "!4", "*6", "!6", "*"] = Nothing + | otherwise = Just configServerHost + filterAddrs xs = + case configServerHost of + "*4" -> ipv4Addrs xs ++ ipv6Addrs xs + "!4" -> ipv4Addrs xs + "*6" -> ipv6Addrs xs ++ ipv4Addrs xs + "!6" -> ipv6Addrs xs + _ -> xs + ipv4Addrs = filter ((/=) AF_INET6 . addrFamily) + ipv6Addrs = filter ((==) AF_INET6 . addrFamily) + + addrs <- getAddrInfo (Just $ defaultHints { addrSocketType = Stream }) (T.unpack <$> host) (Just . show $ configServerPort) + tryAddr `traverse` filterAddrs addrs + where + sendEmpty sock = void $ send sock mempty + tryAddr :: AddrInfo -> IO (Either IOException ()) + tryAddr addr = do + sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr) + try $ do + connect sock $ addrAddress addr + withSocketsDo $ bracket (pure sock) close sendEmpty diff --git a/src/PostgREST/App.hs b/src/PostgREST/App.hs index f44192d73..59fda216a 100644 --- a/src/PostgREST/App.hs +++ b/src/PostgREST/App.hs @@ -33,6 +33,7 @@ import qualified Hasql.Transaction.Sessions as SQL import qualified Network.Wai as Wai import qualified Network.Wai.Handler.Warp as Warp +import qualified PostgREST.Admin as Admin import qualified PostgREST.ApiRequest as ApiRequest import qualified PostgREST.ApiRequest.Types as ApiRequestTypes import qualified PostgREST.AppState as AppState @@ -43,7 +44,6 @@ import qualified PostgREST.Logger as Logger import qualified PostgREST.Plan as Plan import qualified PostgREST.Query as Query import qualified PostgREST.Response as Response -import qualified PostgREST.Workers as Workers import PostgREST.ApiRequest (Action (..), ApiRequest (..), Mutation (..), Target (..)) @@ -68,14 +68,14 @@ type SocketRunner = Warp.Settings -> Wai.Application -> FileMode -> FilePath -> run :: SignalHandlerInstaller -> Maybe SocketRunner -> AppState -> IO () run installHandlers maybeRunWithSocket appState = do conf@AppConfig{..} <- AppState.getConfig appState - Workers.connectionWorker appState -- Loads the initial SchemaCache + AppState.connectionWorker appState -- Loads the initial SchemaCache installHandlers appState -- reload schema cache + config on NOTIFY - Workers.runListener conf appState + AppState.runListener conf appState - Workers.runAdmin conf appState $ serverSettings conf + Admin.runAdmin conf appState $ serverSettings conf - let app = postgrest conf appState (Workers.connectionWorker appState) + let app = postgrest conf appState (AppState.connectionWorker appState) case configServerUnixSocket of Just socket -> diff --git a/src/PostgREST/AppState.hs b/src/PostgREST/AppState.hs index 950440dbb..b51d09b2d 100644 --- a/src/PostgREST/AppState.hs +++ b/src/PostgREST/AppState.hs @@ -1,9 +1,10 @@ +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE RecordWildCards #-} module PostgREST.AppState ( AppState , destroy - , flushPool , getConfig , getSchemaCache , getIsListenerOn @@ -11,40 +12,50 @@ module PostgREST.AppState , getPgVersion , getRetryNextIn , getTime - , getWorkerSem , init , initWithPool , logWithZTime - , logPgrstError - , putConfig , putSchemaCache - , putIsListenerOn , putPgVersion - , putRetryNextIn - , signalListener , usePool - , waitListener , debounceLogAcquisitionTimeout + , loadSchemaCache + , reReadConfig + , connectionWorker + , runListener ) where -import qualified Data.ByteString.Lazy as LBS -import qualified Data.Text.Encoding as T -import qualified Hasql.Pool as SQL -import qualified Hasql.Session as SQL -import qualified PostgREST.Error as Error +import qualified Data.ByteString as BS +import qualified Data.ByteString.Lazy as LBS +import qualified Data.Text.Encoding as T +import Hasql.Connection (acquire) +import qualified Hasql.Notifications as SQL +import qualified Hasql.Pool as SQL +import qualified Hasql.Session as SQL +import qualified Hasql.Transaction.Sessions as SQL +import qualified PostgREST.Error as Error import Control.AutoUpdate (defaultUpdateSettings, mkAutoUpdate, updateAction) import Control.Debounce +import Control.Retry (RetryStatus, capDelay, exponentialBackoff, + retrying, rsPreviousDelay) import Data.IORef (IORef, atomicWriteIORef, newIORef, readIORef) import Data.Time (ZonedTime, defaultTimeLocale, formatTime, getZonedTime) import Data.Time.Clock (UTCTime, getCurrentTime) -import PostgREST.Config (AppConfig (..)) -import PostgREST.Config.PgVersion (PgVersion (..), minimumPgVersion) -import PostgREST.SchemaCache (SchemaCache) +import PostgREST.Config (AppConfig (..), + readAppConfig) +import PostgREST.Config.Database (queryDbSettings, + queryPgVersion, + queryRoleSettings) +import PostgREST.Config.PgVersion (PgVersion (..), + minimumPgVersion) +import PostgREST.SchemaCache (SchemaCache, + querySchemaCache) +import PostgREST.SchemaCache.Identifiers (dumpQi) import Protolude @@ -56,8 +67,8 @@ data AppState = AppState , statePgVersion :: IORef PgVersion -- | No schema cache at the start. Will be filled in by the connectionWorker , stateSchemaCache :: IORef (Maybe SchemaCache) - -- | Binary semaphore to make sure just one connectionWorker can run at a time - , stateWorkerSem :: MVar () + -- | starts the connection worker with a debounce + , debouncedConnectionWorker :: IO () -- | Binary semaphore used to sync the listener(NOTIFY reload) with the connectionWorker. , stateListener :: MVar () -- | State of the LISTEN channel, used for the admin server checks @@ -86,7 +97,7 @@ initWithPool pool conf = do appState <- AppState pool <$> newIORef minimumPgVersion -- assume we're in a supported version when starting, this will be corrected on a later step <*> newIORef Nothing - <*> newEmptyMVar + <*> pure (pure ()) <*> newEmptyMVar <*> newIORef False <*> newIORef conf @@ -96,7 +107,8 @@ initWithPool pool conf = do <*> newIORef 0 <*> pure (pure ()) - deb <- + + debLogTimeout <- let oneSecond = 1000000 in mkDebounce defaultDebounceSettings { debounceAction = logPgrstError appState SQL.AcquisitionTimeoutUsageError @@ -104,7 +116,15 @@ initWithPool pool conf = do , debounceEdge = leadingEdge -- logs at the start and the end } - return appState { debounceLogAcquisitionTimeout = deb } + debWorker <- + let decisecond = 100000 in + mkDebounce defaultDebounceSettings + { debounceAction = internalConnectionWorker appState + , debounceFreq = decisecond + , debounceEdge = leadingEdge -- runs the worker at the start and the end + } + + return appState { debounceLogAcquisitionTimeout = debLogTimeout, debouncedConnectionWorker = debWorker } destroy :: AppState -> IO () destroy = destroyPool @@ -143,8 +163,8 @@ getSchemaCache = readIORef . stateSchemaCache putSchemaCache :: AppState -> Maybe SchemaCache -> IO () putSchemaCache appState = atomicWriteIORef (stateSchemaCache appState) -getWorkerSem :: AppState -> MVar () -getWorkerSem = stateWorkerSem +connectionWorker :: AppState -> IO () +connectionWorker = debouncedConnectionWorker getRetryNextIn :: AppState -> IO Int getRetryNextIn = readIORef . stateRetryNextIn @@ -189,3 +209,236 @@ getIsListenerOn = readIORef . stateIsListenerOn putIsListenerOn :: AppState -> Bool -> IO () putIsListenerOn = atomicWriteIORef . stateIsListenerOn + +-- | Schema cache status +data SCacheStatus + = SCLoaded + | SCOnRetry + | SCFatalFail + +-- | Load the SchemaCache by using a connection from the pool. +loadSchemaCache :: AppState -> IO SCacheStatus +loadSchemaCache appState = do + conf@AppConfig{..} <- getConfig appState + result <- + let transaction = if configDbPreparedStatements then SQL.transaction else SQL.unpreparedTransaction in + usePool appState . transaction SQL.ReadCommitted SQL.Read $ + querySchemaCache conf + case result of + Left e -> do + case Error.checkIsFatal e of + Just hint -> do + logWithZTime appState "A fatal error ocurred when loading the schema cache" + logPgrstError appState e + logWithZTime appState hint + return SCFatalFail + Nothing -> do + putSchemaCache appState Nothing + logWithZTime appState "An error ocurred when loading the schema cache" + logPgrstError appState e + return SCOnRetry + + Right sCache -> do + putSchemaCache appState (Just sCache) + logWithZTime appState "Schema cache loaded" + return SCLoaded + +-- | Current database connection status data ConnectionStatus +data ConnectionStatus + = NotConnected + | Connected PgVersion + | FatalConnectionError Text + deriving (Eq) + +-- | The purpose of this worker is to obtain a healthy connection to pg and an +-- up-to-date schema cache(SchemaCache). This method is meant to be called +-- multiple times by the same thread, but does nothing if the previous +-- invocation has not terminated. In all cases this method does not halt the +-- calling thread, the work is performed in a separate thread. +-- +-- Background thread that does the following : +-- 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 +-- program. +-- 3. Obtains the sCache. If this fails, it goes back to 1. +internalConnectionWorker :: AppState -> IO () +internalConnectionWorker appState = work + where + work = do + AppConfig{..} <- getConfig appState + logWithZTime appState "Attempting to connect to the database..." + connected <- establishConnection appState + case connected of + FatalConnectionError reason -> + -- Fatal error when connecting + logWithZTime appState reason >> killThread (getMainThreadId appState) + NotConnected -> + -- Unreachable because establishConnection will keep trying to connect + return () + Connected actualPgVersion -> do + -- Procede with initialization + putPgVersion appState actualPgVersion + when configDbChannelEnabled $ + signalListener appState + logWithZTime appState "Connection successful" + -- this could be fail because the connection drops, but the loadSchemaCache will pick the error and retry again + -- We cannot retry after it fails immediately, because db-pre-config could have user errors. We just log the error and continue. + when configDbConfig $ reReadConfig False appState + scStatus <- loadSchemaCache appState + case scStatus of + SCLoaded -> + -- do nothing and proceed if the load was successful + return () + SCOnRetry -> + -- retry reloading the schema cache + work + SCFatalFail -> + -- die if our schema cache query has an error + killThread $ getMainThreadId appState + +-- | Repeatedly flush the pool, and check if a connection from the +-- pool allows access to the PostgreSQL database. +-- +-- Releasing the pool is key for rapid recovery. Otherwise, the pool +-- timeout would have to be reached for new healthy connections to be acquired. +-- Which might not happen if the server is busy with requests. No idle +-- connection, no pool timeout. +-- +-- The connection tries are capped, but if the connection times out no error is +-- thrown, just 'False' is returned. +establishConnection :: AppState -> IO ConnectionStatus +establishConnection appState = + retrying retrySettings shouldRetry $ + const $ flushPool appState >> getConnectionStatus + where + retrySettings = capDelay delayMicroseconds $ exponentialBackoff backoffMicroseconds + delayMicroseconds = 32000000 -- 32 seconds + backoffMicroseconds = 1000000 -- 1 second + + getConnectionStatus :: IO ConnectionStatus + getConnectionStatus = do + pgVersion <- usePool appState $ queryPgVersion False -- No need to prepare the query here, as the connection might not be established + case pgVersion of + Left e -> do + logPgrstError appState e + case Error.checkIsFatal e of + Just reason -> + return $ FatalConnectionError reason + Nothing -> + return NotConnected + Right version -> + if version < minimumPgVersion then + return . FatalConnectionError $ + "Cannot run in this PostgreSQL version, PostgREST needs at least " + <> pgvName minimumPgVersion + else + return . Connected $ version + + shouldRetry :: RetryStatus -> ConnectionStatus -> IO Bool + shouldRetry rs isConnSucc = do + let + delay = fromMaybe 0 (rsPreviousDelay rs) `div` backoffMicroseconds + itShould = NotConnected == isConnSucc + when itShould . logWithZTime appState $ + "Attempting to reconnect to the database in " + <> (show delay::Text) + <> " seconds..." + when itShould $ putRetryNextIn appState delay + return itShould + +-- | Re-reads the config plus config options from the db +reReadConfig :: Bool -> AppState -> IO () +reReadConfig startingUp appState = do + AppConfig{..} <- getConfig appState + dbSettings <- + if configDbConfig then do + qDbSettings <- usePool appState $ queryDbSettings (dumpQi <$> configDbPreConfig) configDbPreparedStatements + case qDbSettings of + Left e -> do + logWithZTime appState + "An error ocurred when trying to query database settings for the config parameters" + case Error.checkIsFatal e of + Just hint -> do + logPgrstError appState e + logWithZTime appState hint + killThread (getMainThreadId appState) + Nothing -> do + logPgrstError appState e + pure mempty + Right x -> pure x + else + pure mempty + roleSettings <- + if configDbConfig then do + rSettings <- usePool appState $ queryRoleSettings configDbPreparedStatements + case rSettings of + Left e -> do + logWithZTime appState "An error ocurred when trying to query the role settings" + logPgrstError appState e + pure mempty + Right x -> pure x + else + pure mempty + readAppConfig dbSettings configFilePath (Just configDbUri) roleSettings >>= \case + Left err -> + if startingUp then + panic err -- die on invalid config if the program is starting up + else + logWithZTime appState $ "Failed reloading config: " <> err + Right newConf -> do + putConfig appState newConf + if startingUp then + pass + else + logWithZTime appState "Config reloaded" + + +runListener :: AppConfig -> AppState -> IO () +runListener AppConfig{configDbChannelEnabled} appState = + when configDbChannelEnabled $ listener appState + +-- | Starts a dedicated pg connection to LISTEN for notifications. When a +-- NOTIFY - with an empty payload - is done, it refills the schema +-- cache. It uses the connectionWorker in case the LISTEN connection dies. +listener :: AppState -> IO () +listener appState = do + AppConfig{..} <- getConfig appState + let dbChannel = toS configDbChannel + + -- The listener has to wait for a signal from the connectionWorker. + -- This is because when the connection to the db is lost, the listener also + -- tries to recover the connection, but not with the same pace as the connectionWorker. + -- Not waiting makes stderr quickly fill with connection retries messages from the listener. + waitListener appState + + -- forkFinally allows to detect if the thread dies + void . flip forkFinally (handleFinally dbChannel) $ do + dbOrError <- acquire $ toUtf8 configDbUri + case dbOrError of + Right db -> do + logWithZTime appState $ "Listening for notifications on the " <> dbChannel <> " channel" + putIsListenerOn appState True + SQL.listen db $ SQL.toPgIdentifier dbChannel + SQL.waitForNotifications handleNotification db + _ -> + die $ "Could not listen for notifications on the " <> dbChannel <> " channel" + where + handleFinally dbChannel _ = do + -- if the thread dies, we try to recover + logWithZTime appState $ "Retrying listening for notifications on the " <> dbChannel <> " channel.." + putIsListenerOn appState False + -- assume the pool connection was also lost, call the connection worker + connectionWorker appState + -- retry the listener + listener appState + + handleNotification _ msg + | BS.null msg = cacheReloader + | msg == "reload schema" = cacheReloader + | msg == "reload config" = 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/CLI.hs b/src/PostgREST/CLI.hs index cb5a36b82..f74c98a11 100644 --- a/src/PostgREST/CLI.hs +++ b/src/PostgREST/CLI.hs @@ -21,7 +21,6 @@ import PostgREST.AppState (AppState) import PostgREST.Config (AppConfig (..)) import PostgREST.SchemaCache (querySchemaCache) import PostgREST.Version (prettyVersion) -import PostgREST.Workers (reReadConfig) import qualified PostgREST.App as App import qualified PostgREST.AppState as AppState @@ -43,7 +42,7 @@ main installSignalHandlers runAppWithSocket CLI{cliCommand, cliPath} = do AppState.destroy (\appState -> case cliCommand of CmdDumpConfig -> do - when configDbConfig $ reReadConfig True appState + when configDbConfig $ AppState.reReadConfig True appState putStr . Config.toText =<< AppState.getConfig appState CmdDumpSchema -> putStrLn =<< dumpSchema appState CmdRun -> App.run installSignalHandlers runAppWithSocket appState) @@ -51,15 +50,12 @@ main installSignalHandlers runAppWithSocket CLI{cliCommand, cliPath} = do -- | Dump SchemaCache schema to JSON dumpSchema :: AppState -> IO LBS.ByteString dumpSchema appState = do - AppConfig{..} <- AppState.getConfig appState + conf@AppConfig{..} <- AppState.getConfig appState result <- let transaction = if configDbPreparedStatements then SQL.transaction else SQL.unpreparedTransaction in AppState.usePool appState $ transaction SQL.ReadCommitted SQL.Read $ - querySchemaCache - (toList configDbSchemas) - configDbExtraSearchPath - configDbPreparedStatements + querySchemaCache conf case result of Left e -> do hPutStrLn stderr $ "An error ocurred when loading the schema cache:\n" <> show e diff --git a/src/PostgREST/Config.hs b/src/PostgREST/Config.hs index 7b559d4c1..9de11e999 100644 --- a/src/PostgREST/Config.hs +++ b/src/PostgREST/Config.hs @@ -103,6 +103,7 @@ data AppConfig = AppConfig , configServerUnixSocketMode :: FileMode , configAdminServerPort :: Maybe Int , configRoleSettings :: RoleSettings + , configInternalSCSleep :: Maybe Int32 } data LogLevel = LogCrit | LogError | LogWarn | LogInfo @@ -267,6 +268,7 @@ parser optPath env dbSettings roleSettings = <*> parseSocketFileMode "server-unix-socket-mode" <*> optInt "admin-server-port" <*> pure roleSettings + <*> optInt "internal-schema-cache-sleep" where parseAppSettings :: C.Key -> C.Parser C.Config [(Text, Text)] parseAppSettings key = addFromEnv . fmap (fmap coerceText) <$> C.subassocs key C.value diff --git a/src/PostgREST/SchemaCache.hs b/src/PostgREST/SchemaCache.hs index b86eded9e..be60b371f 100644 --- a/src/PostgREST/SchemaCache.hs +++ b/src/PostgREST/SchemaCache.hs @@ -26,6 +26,8 @@ module PostgREST.SchemaCache , schemaDescription ) where +import Control.Monad.Extra (whenJust) + import qualified Data.Aeson as JSON import qualified Data.HashMap.Strict as HM import qualified Data.HashMap.Strict.InsOrd as HMI @@ -38,6 +40,7 @@ import qualified Hasql.Transaction as SQL import Contravariant.Extras (contrazip2) import Text.InterpolatedString.Perl6 (q) +import PostgREST.Config (AppConfig (..)) import PostgREST.Config.Database (pgVersionStatement) import PostgREST.Config.PgVersion (PgVersion, pgVersion100, pgVersion110, pgVersion120) @@ -103,15 +106,18 @@ data KeyDep -- | A SQL query that can be executed independently type SqlQuery = ByteString -querySchemaCache :: [Schema] -> [Schema] -> Bool -> SQL.Transaction SchemaCache -querySchemaCache schemas extraSearchPath prepared = do +querySchemaCache :: AppConfig -> SQL.Transaction SchemaCache +querySchemaCache AppConfig{..} = do SQL.sql "set local schema ''" -- This voids the search path. The following queries need this for getting the fully qualified name(schema.name) of every db object pgVer <- SQL.statement mempty $ pgVersionStatement prepared tabs <- SQL.statement schemas $ allTables pgVer prepared - keyDeps <- SQL.statement (schemas, extraSearchPath) $ allViewsKeyDependencies prepared + keyDeps <- SQL.statement (schemas, configDbExtraSearchPath) $ allViewsKeyDependencies prepared m2oRels <- SQL.statement mempty $ allM2OandO2ORels pgVer prepared funcs <- SQL.statement schemas $ allFunctions pgVer prepared cRels <- SQL.statement mempty $ allComputedRels prepared + _ <- + let sleepCall = SQL.Statement "select pg_sleep($1)" (param HE.int4) HD.noResult prepared in + whenJust configInternalSCSleep (`SQL.statement` sleepCall) -- only used for testing let tabsWViewsPks = addViewPrimaryKeys tabs keyDeps rels = addInverseRels $ addM2MRels tabsWViewsPks $ addViewM2OAndO2ORels keyDeps m2oRels @@ -121,6 +127,9 @@ querySchemaCache schemas extraSearchPath prepared = do , dbRelationships = getOverrideRelationshipsMap rels cRels , dbRoutines = funcs } + where + schemas = toList configDbSchemas + prepared = configDbPreparedStatements -- | overrides detected relationships with the computed relationships and gets the RelationshipsMap getOverrideRelationshipsMap :: [Relationship] -> [Relationship] -> RelationshipsMap diff --git a/src/PostgREST/Unix.hs b/src/PostgREST/Unix.hs index f45decce9..5cb09f726 100644 --- a/src/PostgREST/Unix.hs +++ b/src/PostgREST/Unix.hs @@ -14,7 +14,6 @@ import System.Posix.Files (setFileMode) import System.Posix.Types (FileMode) import qualified PostgREST.AppState as AppState -import qualified PostgREST.Workers as Workers import Protolude @@ -49,10 +48,10 @@ installSignalHandlers appState = do -- The SIGUSR1 signal updates the internal 'SchemaCache' by running -- 'connectionWorker' exactly as before. - install Signals.sigUSR1 $ Workers.connectionWorker appState + install Signals.sigUSR1 $ AppState.connectionWorker appState -- Re-read the config on SIGUSR2 - install Signals.sigUSR2 $ Workers.reReadConfig False appState + install Signals.sigUSR2 $ AppState.reReadConfig False appState where install signal handler = void $ Signals.installHandler signal (Signals.Catch handler) Nothing diff --git a/src/PostgREST/Workers.hs b/src/PostgREST/Workers.hs deleted file mode 100644 index f3e98a6d8..000000000 --- a/src/PostgREST/Workers.hs +++ /dev/null @@ -1,347 +0,0 @@ -{-# LANGUAGE LambdaCase #-} -{-# LANGUAGE NamedFieldPuns #-} -{-# LANGUAGE RecordWildCards #-} - -module PostgREST.Workers - ( connectionWorker - , reReadConfig - , runListener - , runAdmin - ) where - -import qualified Data.ByteString as BS -import qualified Data.Text as T -import qualified Hasql.Notifications as SQL -import qualified Hasql.Session as SQL -import qualified Hasql.Transaction.Sessions as SQL -import qualified Network.HTTP.Types.Status as HTTP -import qualified Network.Wai as Wai -import qualified Network.Wai.Handler.Warp as Warp - -import Control.Retry (RetryStatus, capDelay, exponentialBackoff, - retrying, rsPreviousDelay) -import Hasql.Connection (acquire) - -import Network.Socket -import Network.Socket.ByteString - -import PostgREST.AppState (AppState) -import PostgREST.Config (AppConfig (..), - readAppConfig) -import PostgREST.Config.Database (queryDbSettings, - queryPgVersion, - queryRoleSettings) -import PostgREST.Config.PgVersion (PgVersion (..), - minimumPgVersion) -import PostgREST.Error (checkIsFatal) -import PostgREST.SchemaCache (querySchemaCache) -import PostgREST.SchemaCache.Identifiers (dumpQi) - -import qualified PostgREST.AppState as AppState - -import Protolude - - --- | Current database connection status data ConnectionStatus -data ConnectionStatus - = NotConnected - | Connected PgVersion - | FatalConnectionError Text - deriving (Eq) - --- | Schema cache status -data SCacheStatus - = SCLoaded - | SCOnRetry - | SCFatalFail - --- | The purpose of this worker is to obtain a healthy connection to pg and an --- up-to-date schema cache(SchemaCache). This method is meant to be called --- multiple times by the same thread, but does nothing if the previous --- invocation has not terminated. In all cases this method does not halt the --- calling thread, the work is performed in a separate thread. --- --- Background thread that does the following : --- 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 --- program. --- 3. Obtains the sCache. If this fails, it goes back to 1. -connectionWorker :: AppState -> IO () -connectionWorker appState = do - runExclusively (AppState.getWorkerSem appState) work - -- Prevents multiple workers to be running at the same time. Could happen on - -- too many SIGUSR1s. - where - runExclusively mvar action = mask_ $ do - success <- tryPutMVar mvar () - when success $ do - void $ forkIO $ action `finally` takeMVar mvar - work = do - AppConfig{..} <- AppState.getConfig appState - AppState.logWithZTime appState "Attempting to connect to the database..." - connected <- establishConnection appState - case connected of - FatalConnectionError reason -> - -- Fatal error when connecting - AppState.logWithZTime appState reason >> killThread (AppState.getMainThreadId appState) - NotConnected -> - -- Unreachable because establishConnection will keep trying to connect - return () - Connected actualPgVersion -> do - -- Procede with initialization - AppState.putPgVersion appState actualPgVersion - when configDbChannelEnabled $ - AppState.signalListener appState - AppState.logWithZTime appState "Connection successful" - -- this could be fail because the connection drops, but the loadSchemaCache will pick the error and retry again - -- We cannot retry after it fails immediately, because db-pre-config could have user errors. We just log the error and continue. - when configDbConfig $ reReadConfig False appState - scStatus <- loadSchemaCache appState - case scStatus of - SCLoaded -> - -- do nothing and proceed if the load was successful - return () - SCOnRetry -> - -- retry reloading the schema cache - work - SCFatalFail -> - -- die if our schema cache query has an error - killThread $ AppState.getMainThreadId appState - --- | Repeatedly flush the pool, and check if a connection from the --- pool allows access to the PostgreSQL database. --- --- Releasing the pool is key for rapid recovery. Otherwise, the pool --- timeout would have to be reached for new healthy connections to be acquired. --- Which might not happen if the server is busy with requests. No idle --- connection, no pool timeout. --- --- The connection tries are capped, but if the connection times out no error is --- thrown, just 'False' is returned. -establishConnection :: AppState -> IO ConnectionStatus -establishConnection appState = - retrying retrySettings shouldRetry $ - const $ AppState.flushPool appState >> getConnectionStatus - where - retrySettings = capDelay delayMicroseconds $ exponentialBackoff backoffMicroseconds - delayMicroseconds = 32000000 -- 32 seconds - backoffMicroseconds = 1000000 -- 1 second - - getConnectionStatus :: IO ConnectionStatus - getConnectionStatus = do - pgVersion <- AppState.usePool appState $ queryPgVersion False -- No need to prepare the query here, as the connection might not be established - case pgVersion of - Left e -> do - AppState.logPgrstError appState e - case checkIsFatal e of - Just reason -> - return $ FatalConnectionError reason - Nothing -> - return NotConnected - Right version -> - if version < minimumPgVersion then - return . FatalConnectionError $ - "Cannot run in this PostgreSQL version, PostgREST needs at least " - <> pgvName minimumPgVersion - else - return . Connected $ version - - shouldRetry :: RetryStatus -> ConnectionStatus -> IO Bool - shouldRetry rs isConnSucc = do - let - delay = fromMaybe 0 (rsPreviousDelay rs) `div` backoffMicroseconds - itShould = NotConnected == isConnSucc - when itShould . AppState.logWithZTime appState $ - "Attempting to reconnect to the database in " - <> (show delay::Text) - <> " seconds..." - when itShould $ AppState.putRetryNextIn appState delay - return itShould - --- | Load the SchemaCache by using a connection from the pool. -loadSchemaCache :: AppState -> IO SCacheStatus -loadSchemaCache appState = do - AppConfig{..} <- AppState.getConfig appState - result <- - let transaction = if configDbPreparedStatements then SQL.transaction else SQL.unpreparedTransaction in - AppState.usePool appState . transaction SQL.ReadCommitted SQL.Read $ - querySchemaCache (toList configDbSchemas) configDbExtraSearchPath configDbPreparedStatements - case result of - Left e -> do - case checkIsFatal e of - Just hint -> do - AppState.logWithZTime appState "A fatal error ocurred when loading the schema cache" - AppState.logPgrstError appState e - AppState.logWithZTime appState hint - return SCFatalFail - Nothing -> do - AppState.putSchemaCache appState Nothing - AppState.logWithZTime appState "An error ocurred when loading the schema cache" - AppState.logPgrstError appState e - return SCOnRetry - - Right sCache -> do - AppState.putSchemaCache appState (Just sCache) - AppState.logWithZTime appState "Schema cache loaded" - return SCLoaded - -runListener :: AppConfig -> AppState -> IO () -runListener AppConfig{configDbChannelEnabled} appState = - when configDbChannelEnabled $ listener appState - --- | Starts a dedicated pg connection to LISTEN for notifications. When a --- NOTIFY - with an empty payload - is done, it refills the schema --- cache. It uses the connectionWorker in case the LISTEN connection dies. -listener :: AppState -> IO () -listener appState = do - AppConfig{..} <- AppState.getConfig appState - let dbChannel = toS configDbChannel - - -- The listener has to wait for a signal from the connectionWorker. - -- This is because when the connection to the db is lost, the listener also - -- tries to recover the connection, but not with the same pace as the connectionWorker. - -- Not waiting makes stderr quickly fill with connection retries messages from the listener. - AppState.waitListener appState - - -- forkFinally allows to detect if the thread dies - void . flip forkFinally (handleFinally dbChannel) $ do - dbOrError <- acquire $ toUtf8 configDbUri - case dbOrError of - Right db -> do - AppState.logWithZTime appState $ "Listening for notifications on the " <> dbChannel <> " channel" - AppState.putIsListenerOn appState True - SQL.listen db $ SQL.toPgIdentifier dbChannel - SQL.waitForNotifications handleNotification db - _ -> - die $ "Could not listen for notifications on the " <> dbChannel <> " channel" - where - handleFinally dbChannel _ = do - -- if the thread dies, we try to recover - AppState.logWithZTime appState $ "Retrying listening for notifications on the " <> dbChannel <> " channel.." - AppState.putIsListenerOn appState False - -- assume the pool connection was also lost, call the connection worker - connectionWorker appState - -- retry the listener - listener appState - - handleNotification _ msg - | BS.null msg = cacheReloader - | msg == "reload schema" = cacheReloader - | msg == "reload config" = 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 - --- | Re-reads the config plus config options from the db -reReadConfig :: Bool -> AppState -> IO () -reReadConfig startingUp appState = do - AppConfig{..} <- AppState.getConfig appState - dbSettings <- - if configDbConfig then do - qDbSettings <- AppState.usePool appState $ queryDbSettings (dumpQi <$> configDbPreConfig) configDbPreparedStatements - case qDbSettings of - Left e -> do - AppState.logWithZTime appState - "An error ocurred when trying to query database settings for the config parameters" - case checkIsFatal e of - Just hint -> do - AppState.logPgrstError appState e - AppState.logWithZTime appState hint - killThread (AppState.getMainThreadId appState) - Nothing -> do - AppState.logPgrstError appState e - pure mempty - Right x -> pure x - else - pure mempty - roleSettings <- - if configDbConfig then do - rSettings <- AppState.usePool appState $ queryRoleSettings configDbPreparedStatements - case rSettings of - Left e -> do - AppState.logWithZTime appState "An error ocurred when trying to query the role settings" - AppState.logPgrstError appState e - pure mempty - Right x -> pure x - else - pure mempty - readAppConfig dbSettings configFilePath (Just configDbUri) roleSettings >>= \case - Left err -> - if startingUp then - panic err -- die on invalid config if the program is starting up - else - AppState.logWithZTime appState $ "Failed reloading config: " <> err - Right newConf -> do - AppState.putConfig appState newConf - if startingUp then - pass - else - AppState.logWithZTime appState "Config reloaded" - -runAdmin :: AppConfig -> AppState -> Warp.Settings -> IO () -runAdmin conf@AppConfig{configAdminServerPort} appState settings = - whenJust configAdminServerPort $ \adminPort -> do - AppState.logWithZTime appState $ "Admin server listening on port " <> show adminPort - void . forkIO $ Warp.runSettings (settings & Warp.setPort adminPort) adminApp - where - whenJust :: Applicative m => Maybe a -> (a -> m ()) -> m () - whenJust mg f = maybe (pure ()) f mg - adminApp = admin appState conf - --- | PostgREST admin application -admin :: AppState.AppState -> AppConfig -> Wai.Application -admin appState appConfig req respond = do - isMainAppReachable <- any isRight <$> reachMainApp appConfig - isSchemaCacheLoaded <- isJust <$> AppState.getSchemaCache appState - isConnectionUp <- - if configDbChannelEnabled appConfig - then AppState.getIsListenerOn appState - else isRight <$> AppState.usePool appState (SQL.sql "SELECT 1") - - case Wai.pathInfo req of - ["ready"] -> - respond $ Wai.responseLBS (if isMainAppReachable && isConnectionUp && isSchemaCacheLoaded then HTTP.status200 else HTTP.status503) [] mempty - ["live"] -> - respond $ Wai.responseLBS (if isMainAppReachable then HTTP.status200 else HTTP.status503) [] mempty - _ -> - respond $ Wai.responseLBS HTTP.status404 [] mempty - --- Try to connect to the main app socket --- Note that it doesn't even send a valid HTTP request, we just want to check that the main app is accepting connections --- The code for resolving the "*4", "!4", "*6", "!6", "*" special values is taken from --- https://hackage.haskell.org/package/streaming-commons-0.2.2.4/docs/src/Data.Streaming.Network.html#bindPortGenEx -reachMainApp :: AppConfig -> IO [Either IOException ()] -reachMainApp AppConfig{..} = - case configServerUnixSocket of - Just path -> do - sock <- socket AF_UNIX Stream 0 - (:[]) <$> try (do - connect sock $ SockAddrUnix path - withSocketsDo $ bracket (pure sock) close sendEmpty) - Nothing -> do - let - host | configServerHost `elem` ["*4", "!4", "*6", "!6", "*"] = Nothing - | otherwise = Just configServerHost - filterAddrs xs = - case configServerHost of - "*4" -> ipv4Addrs xs ++ ipv6Addrs xs - "!4" -> ipv4Addrs xs - "*6" -> ipv6Addrs xs ++ ipv4Addrs xs - "!6" -> ipv6Addrs xs - _ -> xs - ipv4Addrs = filter ((/=) AF_INET6 . addrFamily) - ipv6Addrs = filter ((==) AF_INET6 . addrFamily) - - addrs <- getAddrInfo (Just $ defaultHints { addrSocketType = Stream }) (T.unpack <$> host) (Just . show $ configServerPort) - tryAddr `traverse` filterAddrs addrs - where - sendEmpty sock = void $ send sock mempty - tryAddr :: AddrInfo -> IO (Either IOException ()) - tryAddr addr = do - sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr) - try $ do - connect sock $ addrAddress addr - withSocketsDo $ bracket (pure sock) close sendEmpty diff --git a/test/io/fixtures.sql b/test/io/fixtures.sql index 638466d24..6187d902f 100644 --- a/test/io/fixtures.sql +++ b/test/io/fixtures.sql @@ -142,3 +142,19 @@ returns text as $$ select current_setting('transaction_isolation', true); $$ language sql set default_transaction_isolation = 'REPEATABLE READ'; + +create or replace function create_function() returns void as $_$ + drop function if exists mult_them(int, int); + create or replace function mult_them(a int, b int) returns int as $$ + select a*b; + $$ language sql; + notify pgrst, 'reload schema'; +$_$ language sql security definer; + +create or replace function migrate_function() returns void as $_$ + drop function if exists mult_them(int, int); + create or replace function mult_them(c int, d int) returns int as $$ + select c*d; + $$ language sql; + notify pgrst, 'reload schema'; +$_$ language sql security definer; diff --git a/test/io/test_io.py b/test/io/test_io.py index a6f49b286..17c363758 100644 --- a/test/io/test_io.py +++ b/test/io/test_io.py @@ -551,7 +551,7 @@ def test_pool_size(defaultenv, metapostgrest): def test_pool_acquisition_timeout(defaultenv, metapostgrest): - "Verify that PGRST_DB_POOL_ACQUISITON_TIMEOUT times out when the pool is empty" + "Verify that PGRST_DB_POOL_ACQUISITION_TIMEOUT times out when the pool is empty" env = { **defaultenv, @@ -946,6 +946,44 @@ def test_isolation_level(defaultenv): assert response.text == '"serializable"' +def test_schema_cache_reloading(defaultenv): + "schema cache should reload successfully" + + # If DB_POOL=1, then the second request(/rpc/migrate_function) will just wait(PGRST_DB_POOL_ACQUISITION_TIMEOUT=10) for the schema cache reload to finish. + # This is bc the only pool connection will be busy with the PGRST_INTERNAL_SCHEMA_CACHE_SLEEP(does a pg_sleep) + # So this must be tested with a DB_POOL size of at least 2. That way the second request will pick the other pool connection and proceed. + + env = { + **defaultenv, + "PGRST_INTERNAL_SCHEMA_CACHE_SLEEP": "1", + "PGRST_DB_CHANNEL_ENABLED": "true", + "PGRST_DB_POOL": "2", + } + + internal_sleep = int(env["PGRST_INTERNAL_SCHEMA_CACHE_SLEEP"]) + + with run(env=env, wait_for_readiness=False) as postgrest: + time.sleep(2 * internal_sleep + 0.1) # wait for readiness manually + + response = postgrest.session.post("/rpc/create_function") + assert response.status_code == 204 + + time.sleep( + internal_sleep / 2 + ) # wait to be inside the schema cache reload process + + response = postgrest.session.post("/rpc/migrate_function") + assert response.status_code == 204 + + time.sleep( + 2 * internal_sleep + ) # wait enough time to ensure the schema cache state remains + + response = postgrest.session.get("/rpc/mult_them?c=3&d=4") + assert response.text == "12" + assert response.status_code == 200 + + # TODO: This test fails now because of https://github.com/PostgREST/postgrest/pull/2122 # The stack size of 1K(-with-rtsopts=-K1K) is not enough and this fails with "stack overflow" # A stack size of 200K seems to be enough for succeess diff --git a/test/spec/Main.hs b/test/spec/Main.hs index 1cf46bdde..917a0c1c8 100644 --- a/test/spec/Main.hs +++ b/test/spec/Main.hs @@ -3,8 +3,7 @@ module Main where import qualified Hasql.Pool as P import qualified Hasql.Transaction.Sessions as HT -import Data.Function (id) -import Data.List.NonEmpty (toList) +import Data.Function (id) import Test.Hspec @@ -70,10 +69,8 @@ main = do actualPgVersion <- either (panic . show) id <$> P.use pool (queryPgVersion False) - baseSchemaCache <- - loadSchemaCache pool - (configDbSchemas testCfg) - (configDbExtraSearchPath testCfg) + -- cached schema cache so most tests run fast + baseSchemaCache <- loadSchemaCache pool testCfg let -- For tests that run with the same refSchemaCache @@ -85,10 +82,7 @@ main = do -- For tests that run with a different SchemaCache(depends on configSchemas) appDbs config = do - customSchemaCache <- - loadSchemaCache pool - (configDbSchemas config) - (configDbExtraSearchPath config) + customSchemaCache <- loadSchemaCache pool config appState <- AppState.initWithPool pool config AppState.putPgVersion appState actualPgVersion AppState.putSchemaCache appState (Just customSchemaCache) @@ -265,5 +259,5 @@ main = do describe "Feature.RollbackForcedSpec" Feature.RollbackSpec.forced where - loadSchemaCache pool schemas extraSearchPath = - either (panic.show) id <$> P.use pool (HT.transaction HT.ReadCommitted HT.Read $ querySchemaCache (toList schemas) extraSearchPath True) + loadSchemaCache pool conf = + either (panic.show) id <$> P.use pool (HT.transaction HT.ReadCommitted HT.Read $ querySchemaCache conf) diff --git a/test/spec/SpecHelper.hs b/test/spec/SpecHelper.hs index 3421db894..c90425eed 100644 --- a/test/spec/SpecHelper.hs +++ b/test/spec/SpecHelper.hs @@ -113,6 +113,7 @@ baseCfg = let secret = Just $ encodeUtf8 "reallyreallyreallyreallyverysafe" in , configDbTxRollbackAll = True , configAdminServerPort = Nothing , configRoleSettings = mempty + , configInternalSCSleep = Nothing } testCfg :: AppConfig