add(observation): db pool flushes observation

DISCLAIMER:
This commit was authored entirely by a human without the assistance of LLMs.

Emit a dedicated PoolFlushed observation when the DB pool is released during schema cache reload.

Author: mkleczek

postgrest.cabal

@@ -259,6 +259,7 @@ test-suite spec
                       Feature.Query.UpsertSpec
                       Feature.RollbackSpec
                       Feature.RpcPreRequestGucsSpec
+                      Feature.SchemaCacheSpec
                       SpecHelper
   build-depends:      base              >= 4.9 && < 4.20
                     , aeson             >= 2.0.3 && < 2.3

src/PostgREST/AppState.hs

@@ -219,10 +219,14 @@ usePool AppState{stateObserver=observer, stateMainThreadId=mainThreadId, ..} ses
 
 -- | Flush the connection pool so that any future use of the pool will
 -- use connections freshly established after this call.
+-- | Emits PoolFlushed observation
 flushPool :: AppState -> IO ()
-flushPool AppState{..} = SQL.release statePool
+flushPool AppState{..} = do
+  SQL.release statePool
+  stateObserver PoolFlushed
 
 -- | Destroy the pool on shutdown.
+-- | Differs from flushPool in not emiting PoolFlushed observation.
 destroyPool :: AppState -> IO ()
 destroyPool AppState{..} = SQL.release statePool
 

src/PostgREST/Logger.hs

@@ -111,6 +111,9 @@ observationLogger loggerState logLevel obs = case obs of
   o@PoolRequestFullfilled ->
     when (logLevel >= LogDebug) $ do
       logWithZTime loggerState $ observationMessages o
+  o@PoolFlushed ->
+    when (logLevel >= LogDebug) $ do
+      logWithZTime loggerState $ observationMessages o
   o@JwtCacheEviction ->
     when (logLevel >= LogDebug) $ do
       logWithZTime loggerState $ observationMessages o
@@ -224,6 +227,8 @@ observationMessages = \case
     pure "Trying to borrow a connection from pool"
   PoolRequestFullfilled ->
     pure "Borrowed a connection from the pool"
+  PoolFlushed ->
+    pure "Database connection pool flushed"
   JwtCacheLookup _ ->
     pure "Looked up a JWT in JWT cache"
   JwtCacheEviction ->

src/PostgREST/Observation.hs

@@ -53,6 +53,7 @@ data Observation
   | HasqlPoolObs SQL.Observation
   | PoolRequest
   | PoolRequestFullfilled
+  | PoolFlushed
   | JwtCacheLookup Bool
   | JwtCacheEviction
   | WarpErrorObs Text

test/spec/Feature/SchemaCacheSpec.hs

@@ -0,0 +1,51 @@
+{-# LANGUAGE DataKinds           #-}
+{-# LANGUAGE MonadComprehensions #-}
+{-# LANGUAGE NamedFieldPuns      #-}
+module Feature.SchemaCacheSpec where
+
+import           Network.Wai           (Application)
+import qualified PostgREST.AppState    as AppState
+import           PostgREST.Config      (configDbSchemas)
+import           PostgREST.Observation
+import           Protolude
+import           SpecHelper
+import           Test.Hspec            (SpecWith, describe, it)
+import           Test.Hspec.Wai        (getState)
+
+spec :: SpecWith (SpecState, Application)
+spec = describe "Server started with metrics enabled" $ do
+
+  it "Should emit PoolFlushed, SchemaCacheQueriedObs and SchemaCacheLoadedObs when schema cache is reloaded" $ do
+    SpecState{specAppState = appState, specObsChan} <- getState
+    let waitFor = waitForObs specObsChan
+
+    liftIO $ do
+      AppState.schemaCacheLoader appState
+
+      waitFor (1 * sec) "PoolFlushed" $ \x -> [ o | o@PoolFlushed <- pure x ]
+      waitFor (1 * sec) "SchemaCacheQueriedObs" $ \x -> [ o | o@SchemaCacheQueriedObs{} <- pure x ]
+      waitFor (1 * sec) "SchemaCacheLoadedObs" $ \x -> [ o | o@SchemaCacheLoadedObs{} <- pure x ]
+
+
+  it "Should flush pool multiple times when schema reloading retries" $ do
+    SpecState{specAppState = appState, specObsChan} <- getState
+    let waitFor = waitForObs specObsChan
+
+    liftIO $ do
+        AppState.getConfig appState >>= \cfg -> do
+          AppState.putConfig appState $ cfg { configDbSchemas = pure "bad_schema" }
+          AppState.schemaCacheLoader appState
+
+          waitFor (1 * sec) "PoolFlushed 1" $ \x -> [ o | o@PoolFlushed <- pure x ]
+          waitFor (1 * sec) "SchemaCacheErrorObs" $ \x -> [ o | o@SchemaCacheErrorObs{} <- pure x ]
+
+          -- Restore configuration
+          AppState.putConfig appState cfg
+
+        -- Wait for 2 seconds so that retry can happen
+        waitFor (2 * sec) "PoolFlushed 2" $ \x -> [ o | o@PoolFlushed <- pure x ]
+        waitFor (1 * sec) "SchemaCacheQueriedObs" $ \x -> [ o | o@SchemaCacheQueriedObs{} <- pure x ]
+        waitFor (1 * sec) "SchemaCacheLoadedObs" $ \x -> [ o | o@SchemaCacheLoadedObs{} <- pure x ]
+
+  where
+    sec = 1000000

test/spec/Main.hs

@@ -69,6 +69,7 @@ import qualified Feature.Query.UpdateSpec
 import qualified Feature.Query.UpsertSpec
 import qualified Feature.RollbackSpec
 import qualified Feature.RpcPreRequestGucsSpec
+import qualified Feature.SchemaCacheSpec
 import           PostgREST.Observation                 (Observation (HasqlPoolObs))
 
 
@@ -300,6 +301,9 @@ main = do
     before observationsApp $
       describe "Feature.MetricsSpec" Feature.MetricsSpec.spec
 
+    before observationsApp $
+      describe "Feature.SchemaCacheSpec" Feature.SchemaCacheSpec.spec
+
   where
     loadSCache pool conf =
       either (panic.show) id <$> P.use pool (HT.transaction HT.ReadCommitted HT.Read $ querySchemaCache conf)