fix: Flush pool as late as possible during schema cache reloading

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

retryingSchemaCacheLoad flushes the pool upon every retry before it starts reloading the schema. This is too early as schema reloading might take some time during which new connections might be acquired. The consequence is that:
* upon successful schema cache reload we might have some connections created with the old schema cache
* we close connections upon each retry and under load we will keep closing and re-opening connections until schema cache load succeeds

This change is to make sure we flush the pool only after successful schema cache querying but before loading (so that connections acquired during loading wait for it and do not interfere with timing the loading process).

Author: mkleczek

src/PostgREST/AppState.hs

@@ -309,8 +309,6 @@ retryingSchemaCacheLoad appState@AppState{stateObserver=observer, stateMainThrea
       let delay = fromMaybe 0 rsPreviousDelay `div` oneSecondInUs
       observer $ ConnectionRetryObs delay
 
-    flushPool appState
-
     (,) <$> qPgVersion <*> (qInDbConfig *> qSchemaCache)
   )
   where
@@ -359,6 +357,10 @@ retryingSchemaCacheLoad appState@AppState{stateObserver=observer, stateMainThrea
           -- IORef on putSchemaCache. This is why schema cache status is marked as pending here to signal the Admin server (using isPending) that we're on a recovery state.
           markSchemaCachePending appState
           putSchemaCache appState $ Just sCache
+          -- Flush the pool after loading the schema cache to reset any stale session cache entries
+          -- We do it after successfully querying the schema cache
+          -- and after marking sCacheStatus as pending,
+          flushPool appState
           observer $ SchemaCacheQueriedObs resultTime
           observer . uncurry SchemaCacheLoadedObs =<< timeItT (evaluate $ showSummary sCache)
           markSchemaCacheLoaded appState

test/io/test_io.py

@@ -714,7 +714,7 @@ def test_log_level(level, defaultenv):
         response = postgrest.session.get("/")
         assert response.status_code == 200
 
-        output = postgrest.read_stdout(nlines=7)
+        output = postgrest.read_stdout(nlines=9)
 
         if level == "crit":
             assert len(output) == 0
@@ -752,7 +752,7 @@ def test_log_level(level, defaultenv):
                     r'- - postgrest_test_anonymous \[.+\] "GET / HTTP/1.1" 200 \d+ "" "python-requests/.+"',
                 ],
             )
-            assert len(output) == 7
+            assert len(output) == 9
             assert any("Connection" and "is available" in line for line in output)
             assert any("Connection" and "is used" in line for line in output)
 
@@ -1389,7 +1389,7 @@ def test_db_error_logging_to_stderr(level, defaultenv, metapostgrest):
         assert response.status_code == 500
 
         # ensure the message appears on the logs
-        output = postgrest.read_stdout(nlines=6)
+        output = postgrest.read_stdout(nlines=8)
 
         if level == "crit":
             assert len(output) == 0
@@ -1606,7 +1606,7 @@ def test_log_pool_req_observation(level, defaultenv):
 
         if level == "debug":
             output = postgrest.read_stdout(nlines=7)
-            assert len(output) == 6
+            assert len(output) == 7
             match_log(output, [pool_req, pool_req_fullfill])
         elif level == "info":
             output = postgrest.read_stdout(nlines=4)

test/spec/Feature/SchemaCacheSpec.hs

@@ -0,0 +1,49 @@
+{-# 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 once 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) "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" $ \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