diff --git a/.circleci/config.yml b/.circleci/config.yml index 5e5eac298..7d76ca655 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -267,6 +267,8 @@ workflows: filters: tags: only: /v[0-9]+(\.[0-9]+)*/ + requires: + - nix-build - release: requires: - style-check diff --git a/CHANGELOG.md b/CHANGELOG.md index 87d901ae3..67d3ae6c9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ This project adheres to [Semantic Versioning](http://semver.org/). ### Added - #1525, Allow http status override through response.status guc - @steve-chavez + - #1512, Allow schema cache reloading with NOTIFY - @steve-chavez ### Fixed diff --git a/main/Main.hs b/main/Main.hs index d89ad3b36..f3a48bf40 100644 --- a/main/Main.hs +++ b/main/Main.hs @@ -4,11 +4,17 @@ module Main where import qualified Data.ByteString as BS import qualified Data.ByteString.Base64 as B64 +import qualified Hasql.Connection as C +import qualified Hasql.Notifications as N import qualified Hasql.Pool as P import qualified Hasql.Transaction.Sessions as HT import Control.AutoUpdate (defaultUpdateSettings, mkAutoUpdate, updateAction) +import Control.Debounce (debounceAction, debounceEdge, + debounceFreq, + defaultDebounceSettings, mkDebounce, + trailingEdge) import Control.Retry (RetryStatus, capDelay, exponentialBackoff, retrying, rsPreviousDelay) @@ -42,6 +48,12 @@ import System.Posix.Signals import UnixSocket #endif +-- Time constants +_32s :: Int +_32s = 32000000 :: Int -- 32 seconds + +_1s :: Int +_1s = 1000000 :: Int -- 1 second {-| The purpose of this worker is to fill the refDbStructure created in 'main' @@ -58,19 +70,18 @@ import UnixSocket 2. Checks if the pg version is supported and if it's not it kills the main program. 3. Obtains the dbStructure. - 4. If 2 or 3 fail to give their result it means the connection is down so it - goes back to 1, otherwise it finishes his work successfully. -} connectionWorker - :: ThreadId -- ^ This thread is killed if pg version is unsupported + :: ThreadId -- ^ Main thread id. Killed if pg version is unsupported -> P.Pool -- ^ The PostgreSQL connection pool -> [Schema] -- ^ Schemas PostgREST is serving up -> IORef (Maybe DbStructure) -- ^ mutable reference to 'DbStructure' -> IORef Bool -- ^ Used as a binary Semaphore + -> (Bool, MVar ConnectionStatus) -- ^ For interacting with the LISTEN channel -> IO () -connectionWorker mainTid pool schemas refDbStructure refIsWorkerOn = do +connectionWorker mainTid pool schemas refDbStructure refIsWorkerOn (dbChannelEnabled, mvarConnectionStatus) = do isWorkerOn <- readIORef refIsWorkerOn - unless isWorkerOn $ do + unless isWorkerOn $ do -- Prevents multiple workers to be running at the same time. Could happen on too many SIGUSR1s. atomicWriteIORef refIsWorkerOn True void $ forkIO work where @@ -78,23 +89,29 @@ connectionWorker mainTid pool schemas refDbStructure refIsWorkerOn = do atomicWriteIORef refDbStructure Nothing putStrLn ("Attempting to connect to the database..." :: Text) connected <- connectionStatus pool + when dbChannelEnabled $ + void $ tryPutMVar mvarConnectionStatus connected -- tryPutMVar doesn't lock the thread. It should always succeed since the worker is the only producer. case connected of - FatalConnectionError reason -> hPutStrLn stderr reason - >> killThread mainTid -- Fatal error when connecting - NotConnected -> return () -- Unreachable - Connected actualPgVersion -> do -- Procede with initialization - result <- P.use pool $ do - dbStructure <- HT.transaction HT.ReadCommitted HT.Read $ getDbStructure schemas actualPgVersion - liftIO $ atomicWriteIORef refDbStructure $ Just dbStructure - case result of - Left e -> do - putStrLn ("Failed to query the database. Retrying." :: Text) - hPutStrLn stderr . toS . errorPayload $ PgError False e - work + FatalConnectionError reason -> hPutStrLn stderr reason >> killThread mainTid -- Fatal error when connecting + NotConnected -> return () -- Unreachable because connectionStatus will keep trying to connect + Connected actualPgVersion -> do -- Procede with initialization + putStrLn ("Connection successful" :: Text) + fillSchemaCache pool actualPgVersion schemas refDbStructure + liftIO $ atomicWriteIORef refIsWorkerOn False - Right _ -> do - atomicWriteIORef refIsWorkerOn False - putStrLn ("Connection successful" :: Text) +fillSchemaCache :: P.Pool -> PgVersion -> [Schema] -> IORef (Maybe DbStructure) -> IO () +fillSchemaCache pool actualPgVersion schemas refDbStructure = do + result <- P.use pool $ HT.transaction HT.ReadCommitted HT.Read $ getDbStructure schemas actualPgVersion + case result of + Left e -> do + -- If this error happens it would mean the connection is down again. Improbable because connectionStatus ensured the connection. + -- It's not a problem though, because App.postgrest would retry the connectionWorker or the user can do a SIGSUR1 again. + hPutStrLn stderr . toS . errorPayload $ PgError False e + putStrLn ("Failed to load the schema cache" :: Text) + + Right dbStructure -> do + atomicWriteIORef refDbStructure $ Just dbStructure + putStrLn ("Schema cache loaded" :: Text) {-| Used by 'connectionWorker' to check if the provided db-uri lets @@ -107,7 +124,7 @@ connectionWorker mainTid pool schemas refDbStructure refIsWorkerOn = do -} connectionStatus :: P.Pool -> IO ConnectionStatus connectionStatus pool = - retrying (capDelay 32000000 $ exponentialBackoff 1000000) + retrying (capDelay _32s $ exponentialBackoff _1s) shouldRetry (const $ P.release pool >> getConnectionStatus) where @@ -129,16 +146,50 @@ connectionStatus pool = shouldRetry :: RetryStatus -> ConnectionStatus -> IO Bool shouldRetry rs isConnSucc = do - let delay = fromMaybe 0 (rsPreviousDelay rs) `div` 1000000 + let delay = fromMaybe 0 (rsPreviousDelay rs) `div` _1s itShould = NotConnected == isConnSucc when itShould $ putStrLn $ "Attempting to reconnect to the database in " <> (show delay::Text) <> " seconds..." return itShould - {-| - This is where everything starts. + Starts a dedicated pg connection to LISTEN for notifications. + When a NOTIFY channel(with an empty payload) is done, it refills the schema cache. + It uses the connectionWorker in case the LISTEN connection dies. -} +listener :: ByteString -> Text -> P.Pool -> [Schema] -> IORef (Maybe DbStructure) -> MVar ConnectionStatus -> IO () -> IO () +listener dbUri dbChannel pool schemas refDbStructure mvarConnectionStatus connWorker = start + where + start = do + connStatus <- takeMVar mvarConnectionStatus -- takeMVar makes the thread wait if the MVar is empty(until there's a connection). + case connStatus of + Connected actualPgVersion -> void $ forkFinally (do -- forkFinally allows to detect if the thread dies + dbOrError <- C.acquire dbUri + -- Debounce in case too many NOTIFYs arrive. Could happen on a migration(assuming a pg EVENT TRIGGER is set up). + scFiller <- mkDebounce (defaultDebounceSettings { + debounceAction = fillSchemaCache pool actualPgVersion schemas refDbStructure, + debounceEdge = trailingEdge, -- wait until the function hasn’t been called in _1s + debounceFreq = _1s }) + case dbOrError of + Right db -> do + putStrLn $ "Listening for notifications on the " <> dbChannel <> " channel" + let channelToListen = N.toPgIdentifier dbChannel + N.listen db channelToListen + N.waitForNotifications (\_ msg -> + if BS.null msg + then scFiller -- reload the schema cache + else pure ()) db -- Do nothing if anything else than an empty message is sent + _ -> die errorMessage) + (\_ -> do -- if the thread dies, we try to recover + putStrLn retryMessage + connWorker -- assume the pool connection was also lost, call the connection worker + start) -- retry the listener + _ -> + putStrLn errorMessage -- Should be unreachable. connectionStatus will retry until there's a connection. + errorMessage = "Could not listen for notifications on the " <> dbChannel <> " channel" :: Text + retryMessage = "Retrying listening for notifications on the " <> dbChannel <> " channel.." :: Text + +-- | This is where everything starts. main :: IO () main = do -- @@ -159,7 +210,8 @@ main = do proxy = configOpenAPIProxyUri conf maybeSocketAddr = configSocket conf socketFileMode = configSocketMode conf - pgSettings = toS (configDatabase conf) -- is the db-uri + dbUri = toS (configDbUri conf) + (dbChannelEnabled, dbChannel) = (configDbChannelEnabled conf, toS $ configDbChannel conf) roleClaimKey = configRoleClaimKey conf appSettings = setHost ((fromString . toS) host) -- Warp settings @@ -167,7 +219,6 @@ main = do . setServerName (toS $ "postgrest/" <> prettyVersion) $ defaultSettings - whenLeft socketFileMode panic -- Checks that the provided proxy uri is formated correctly @@ -181,32 +232,32 @@ main = do -- create connection pool with the provided settings, returns either -- a 'Connection' or a 'ConnectionError'. Does not throw. - pool <- P.acquire (configPool conf, configPoolTimeout' conf, pgSettings) - -- + pool <- P.acquire (configPool conf, configPoolTimeout' conf, dbUri) + + -- Used to sync the listener with the connectionWorker. No connection for the listener at first. Only used if dbChannelEnabled=true. + mvarConnectionStatus <- newEmptyMVar + -- To be filled in by connectionWorker refDbStructure <- newIORef Nothing - -- + -- Helper ref to make sure just one connectionWorker can run at a time refIsWorkerOn <- newIORef False - -- + -- This is passed to the connectionWorker method so it can kill the main -- thread if the PostgreSQL's version is not supported. mainTid <- myThreadId - -- - -- Sets the refDbStructure - connectionWorker - mainTid - pool - schemas - refDbStructure - refIsWorkerOn - -- + + let connWorker = connectionWorker mainTid pool schemas refDbStructure refIsWorkerOn (dbChannelEnabled, mvarConnectionStatus) + + -- Sets the initial refDbStructure + connWorker + -- Only for systems with signals: -- -- releases the connection pool whenever the program is terminated, - -- see issue #268 + -- see https://github.com/PostgREST/postgrest/issues/268 -- - -- Plus the SIGHUP signal updates the internal 'DbStructure' by running + -- Plus the SIGUSR1 signal updates the internal 'DbStructure' by running -- 'connectionWorker' exactly as before. #ifndef mingw32_HOST_OS forM_ [sigINT, sigTERM] $ \sig -> @@ -216,15 +267,13 @@ main = do ) Nothing void $ installHandler sigUSR1 ( - Catch $ connectionWorker - mainTid - pool - schemas - refDbStructure - refIsWorkerOn + Catch connWorker ) Nothing #endif + -- reload schema cache on NOTIFY + when dbChannelEnabled $ + listener dbUri dbChannel pool schemas refDbStructure mvarConnectionStatus connWorker -- ask for the OS time at most once per second getTime <- mkAutoUpdate defaultUpdateSettings {updateAction = getCurrentTime} @@ -235,12 +284,7 @@ main = do refDbStructure pool getTime - (connectionWorker - mainTid - pool - schemas - refDbStructure - refIsWorkerOn) + connWorker -- run the postgrest application with user defined socket. Only for UNIX systems. #ifndef mingw32_HOST_OS @@ -311,14 +355,14 @@ loadSecretFile conf = extractAndTransform mSecret loadDbUriFile :: AppConfig -> IO AppConfig loadDbUriFile conf = extractDbUri mDbUri where - mDbUri = configDatabase conf + mDbUri = configDbUri conf extractDbUri :: Text -> IO AppConfig extractDbUri dbUri = fmap setDbUri $ case stripPrefix "@" dbUri of Nothing -> return dbUri Just filename -> strip <$> readFile (toS filename) - setDbUri dbUri = conf {configDatabase = dbUri} + setDbUri dbUri = conf {configDbUri = dbUri} -- Utilitarian functions. whenJust :: Applicative f => Maybe a -> (a -> f ()) -> f () diff --git a/nix/docker/default.nix b/nix/docker/default.nix index 913e39eb6..e71590323 100644 --- a/nix/docker/default.nix +++ b/nix/docker/default.nix @@ -29,6 +29,8 @@ let "PGRST_DB_POOL=100" "PGRST_DB_POOL_TIMEOUT=10" "PGRST_DB_EXTRA_SEARCH_PATH=public" + "PGRST_DB_CHANNEL=pgrst" + "PGRST_DB_CHANNEL_ENABLED=false" "PGRST_SERVER_HOST=*4" "PGRST_SERVER_PORT=3000" "PGRST_OPENAPI_SERVER_PROXY_URI=" diff --git a/nix/docker/postgrest.conf b/nix/docker/postgrest.conf index 6e22c38f1..82f022dfc 100644 --- a/nix/docker/postgrest.conf +++ b/nix/docker/postgrest.conf @@ -12,6 +12,9 @@ db-pool = "$(PGRST_DB_POOL)" db-pool-timeout = "$(PGRST_DB_POOL_TIMEOUT)" db-extra-search-path = "$(PGRST_DB_EXTRA_SEARCH_PATH)" +db-channel = "$(PGRST_DB_CHANNEL)" +db-channel-enabled = "$(PGRST_DB_CHANNEL_ENABLED)" + server-host = "$(PGRST_SERVER_HOST)" server-port = "$(PGRST_SERVER_PORT)" diff --git a/nix/overlays/haskell-packages/default.nix b/nix/overlays/haskell-packages/default.nix index 943b27cb8..8588e3e35 100644 --- a/nix/overlays/haskell-packages/default.nix +++ b/nix/overlays/haskell-packages/default.nix @@ -12,6 +12,19 @@ let ver = "0.3.0"; sha256 = "0iwh4wsjhb7pms88lw1afhdal9f86nrrkkvv65f9wxbd1b159n72"; } { }; + # To get the sha256 + # nix-prefetch-url --unpack https://hackage.haskell.org/package/hasql-notifications-0.1.0.0/hasql-notifications-0.1.0.0.tar.gz + hasql-notifications = + self.haskell.lib.overrideCabal + ( + prev.callHackageDirect + { + pkg = "hasql-notifications"; + ver = "0.1.0.0"; + sha256 = "1z17gsqvvzzi0yipc3qy3jz8vzpww4vsc4vaj2kbzr2mfliq6fx3"; + } { } + ) + (old: { doCheck = false; }); } // extraOverrides final prev; in { diff --git a/nix/patches/default.nix b/nix/patches/default.nix index 97d470b25..c60a0e292 100644 --- a/nix/patches/default.nix +++ b/nix/patches/default.nix @@ -31,4 +31,7 @@ # https://github.com/nh2/static-haskell-nix/pull/91 static-haskell-nix-postgrest-openssl-linking-fix = ./static-haskell-nix-postgrest-openssl-linking-fix.patch; + + static-haskell-nix-hasql-notifications-openssl-linking-fix = + ./static-haskell-nix-hasql-notifications-openssl-linking-fix.patch; } diff --git a/nix/patches/static-haskell-nix-hasql-notifications-openssl-linking-fix.patch b/nix/patches/static-haskell-nix-hasql-notifications-openssl-linking-fix.patch new file mode 100644 index 000000000..d4bf09e4a --- /dev/null +++ b/nix/patches/static-haskell-nix-hasql-notifications-openssl-linking-fix.patch @@ -0,0 +1,28 @@ +From 49ecb703d9d0bfd38eb69ba5cb63a8262bd03f96 Mon Sep 17 00:00:00 2001 +From: steve-chavez +Date: Thu, 11 Jun 2020 13:18:07 -0500 +Subject: [PATCH] Add hasql-notifications openssl linking fix + +--- + survey/default.nix | 5 +++++ + 1 file changed, 5 insertions(+) + +diff --git a/survey/default.nix b/survey/default.nix +index 828beaa..9c2d5f6 100644 +--- a/survey/default.nix ++++ b/survey/default.nix +@@ -1054,6 +1054,11 @@ let + super.squeal-postgresql + [ final.openssl ] + "--libs openssl"; ++ hasql-notifications = ++ addStaticLinkerFlagsWithPkgconfig ++ super.hasql-notifications ++ [ final.openssl ] ++ "--libs openssl"; + + xml-to-json = + addStaticLinkerFlagsWithPkgconfig +-- +2.19.3 + diff --git a/nix/static-haskell-package.nix b/nix/static-haskell-package.nix index cfb4dc633..31d27089d 100644 --- a/nix/static-haskell-package.nix +++ b/nix/static-haskell-package.nix @@ -18,6 +18,7 @@ let static-haskell-nix [ patches.static-haskell-nix-postgrest-openssl-linking-fix + patches.static-haskell-nix-hasql-notifications-openssl-linking-fix ]; patchedNixpkgs = diff --git a/postgrest.cabal b/postgrest.cabal index 28f2ca06f..8a0f06a11 100644 --- a/postgrest.cabal +++ b/postgrest.cabal @@ -112,6 +112,7 @@ executable postgrest , hasql >= 1.4 && < 1.5 , hasql-pool >= 0.5 && < 0.6 , hasql-transaction >= 0.7.2 && < 1.1 + , hasql-notifications == 0.1.0.0 , network < 3.2 , postgrest , protolude >= 0.3 && < 0.4 diff --git a/src/PostgREST/Config.hs b/src/PostgREST/Config.hs index 903c50b03..f1a9a2c1f 100644 --- a/src/PostgREST/Config.hs +++ b/src/PostgREST/Config.hs @@ -67,7 +67,7 @@ import Protolude.Conv (toS) -- | Config file settings for the server data AppConfig = AppConfig { - configDatabase :: Text + configDbUri :: Text , configAnonRole :: Text , configOpenAPIProxyUri :: Maybe Text , configSchemas :: NonEmpty Text @@ -75,6 +75,8 @@ data AppConfig = AppConfig { , configPort :: Int , configSocket :: Maybe FilePath , configSocketMode :: Either Text FileMode + , configDbChannel :: Text + , configDbChannelEnabled :: Bool , configJwtSecret :: Maybe B.ByteString , configJwtSecretIsBase64 :: Bool @@ -163,6 +165,8 @@ readOptions = do <*> (fromMaybe 3000 <$> optInt "server-port") <*> (fmap unpack <$> optString "server-unix-socket") <*> parseSocketFileMode "server-unix-socket-mode" + <*> (fromMaybe "pgrst" <$> optString "db-channel") + <*> ((Just True ==) <$> optBool "db-channel-enabled") <*> (fmap encodeUtf8 <$> optString "jwt-secret") <*> ((Just True ==) <$> optBool "secret-is-base64") <*> parseJwtAudience "jwt-aud" @@ -276,6 +280,11 @@ readOptions = do |## when none is provided, 660 is applied by default |# server-unix-socket-mode = "660" | + |## Notification channel for reloading the schema cache + |# db-channel = "pgrst" + |## Enable or disable the notification channel + |# db-channel-enabled = false + | |## base url for swagger output |# openapi-server-proxy-uri = "" | diff --git a/stack.yaml b/stack.yaml index a7e770343..e2afa9718 100644 --- a/stack.yaml +++ b/stack.yaml @@ -12,3 +12,4 @@ extra-deps: - hspec-wai-json-0.10.1@sha256:67b405c38f0a9e2771480c8d3ecd8aeb8d8776a35d3b2906cb1b76c9538617e4,1629 - interpolatedstring-perl6-1.0.2@sha256:7ce49c8a69a2a1b89c001ed79db2aab656ffd0faf2a7a701a553b6deb5c8ba7f,1073 - protolude-0.3.0@sha256:8361b811b420585b122a7ba715aa5923834db6e8c36309bf267df2dbf66b95ef,2693 +- hasql-notifications-0.1.0.0@sha256:9ab112d2bb5da0d55abd65f0d27a7bb1dc4aeb792518d9a2ea8a16e243e19985,2156 diff --git a/test/SpecHelper.hs b/test/SpecHelper.hs index 1a12df623..260e07603 100644 --- a/test/SpecHelper.hs +++ b/test/SpecHelper.hs @@ -69,6 +69,10 @@ _baseCfg = -- Connection Settings Nothing -- No user configured Unix Socket file mode (defaults to 660) (Right 432) + -- db-channel + "pgrst" + -- db-channel-enabled + False -- Jwt settings (Just $ encodeUtf8 "reallyreallyreallyreallyverysafe") False Nothing -- Connection Modifiers @@ -88,7 +92,7 @@ _baseCfg = -- Connection Settings [] testCfg :: Text -> AppConfig -testCfg testDbConn = _baseCfg { configDatabase = testDbConn } +testCfg testDbConn = _baseCfg { configDbUri = testDbConn } testCfgNoJWT :: Text -> AppConfig testCfgNoJWT testDbConn = (testCfg testDbConn) { configJwtSecret = Nothing }