fix: deregister relevant connection pool metrics when part of the source config is modified

PR-URL: https://github.com/hasura/graphql-engine-mono/pull/10786
GitOrigin-RevId: 6ba7a1db1df19453c59de3f41d0e0d9de93d3a8e

docker-compose/databases.yaml

@@ -22,6 +22,83 @@ services:
       timeout: 10s
       retries: 20
 
+  # setup for read replicas
+  postgresql-primary:
+    image: 'bitnami/postgresql:16'
+    ports:
+      - 5432
+    volumes:
+      - '/bitnami/postgresql'
+    environment:
+      - POSTGRESQL_REPLICATION_MODE=master
+      - POSTGRESQL_REPLICATION_USER=repl_user
+      - POSTGRESQL_REPLICATION_PASSWORD=repl_password
+      - POSTGRESQL_USERNAME=hasura
+      - POSTGRESQL_PASSWORD=hasura
+      - POSTGRESQL_DATABASE=hasura
+      - ALLOW_EMPTY_PASSWORD=yes
+    healthcheck:
+      test:
+        - CMD-SHELL
+        - export PGPASSWORD="$${POSTGRESQL_PASSWORD:-password}"
+        - psql -U "$${POSTGRESQL_USERNAME:-postgres}" < /dev/null && sleep 5 && psql -U "$${POSTGRESQL_USERNAME:-postgres}" < /dev/null
+      start_period: 5s
+      interval: 5s
+      timeout: 10s
+      retries: 20
+
+  postgresql-replica-1:
+    image: 'bitnami/postgresql:16'
+    ports:
+      - 5432
+    depends_on:
+      postgresql-primary:
+        condition: service_healthy
+    environment:
+      - POSTGRESQL_REPLICATION_MODE=slave
+      - POSTGRESQL_REPLICATION_USER=repl_user
+      - POSTGRESQL_REPLICATION_PASSWORD=repl_password
+      - POSTGRESQL_MASTER_HOST=postgresql-primary
+      - POSTGRESQL_USERNAME=hasura
+      - POSTGRESQL_PASSWORD=hasura
+      - POSTGRESQL_MASTER_PORT_NUMBER=5432
+      - ALLOW_EMPTY_PASSWORD=yes
+    healthcheck:
+      test:
+        - CMD-SHELL
+        - export PGPASSWORD="$${POSTGRESQL_PASSWORD:-password}"
+        - psql -U "$${POSTGRESQL_USERNAME:-postgres}" < /dev/null && sleep 5 && psql -U "$${POSTGRESQL_USERNAME:-postgres}" < /dev/null
+      start_period: 5s
+      interval: 5s
+      timeout: 10s
+      retries: 20
+
+  postgresql-replica-2:
+    image: 'bitnami/postgresql:16'
+    ports:
+      - 5432
+    depends_on:
+      postgresql-primary:
+        condition: service_healthy
+    environment:
+      - POSTGRESQL_REPLICATION_MODE=slave
+      - POSTGRESQL_REPLICATION_USER=repl_user
+      - POSTGRESQL_REPLICATION_PASSWORD=repl_password
+      - POSTGRESQL_MASTER_HOST=postgresql-primary
+      - POSTGRESQL_USERNAME=hasura
+      - POSTGRESQL_PASSWORD=hasura
+      - POSTGRESQL_MASTER_PORT_NUMBER=5432
+      - ALLOW_EMPTY_PASSWORD=yes
+    healthcheck:
+      test:
+        - CMD-SHELL
+        - export PGPASSWORD="$${POSTGRESQL_PASSWORD:-password}"
+        - psql -U "$${POSTGRESQL_USERNAME:-postgres}" < /dev/null && sleep 5 && psql -U "$${POSTGRESQL_USERNAME:-postgres}" < /dev/null
+      start_period: 5s
+      interval: 5s
+      timeout: 10s
+      retries: 20
+
   citus:
     image: citusdata/citus:11.3.0
     platform: linux/amd64

server/lib/test-harness/src/Harness/Constants.hs

@@ -8,6 +8,9 @@ module Harness.Constants
     postgresDb,
     postgresHost,
     postgresPort,
+    postgresPrimaryPort,
+    postgresReplica1Port,
+    postgresReplica2Port,
     postgresqlMetadataConnectionString,
     postgresLivenessCheckAttempts,
     postgresLivenessCheckIntervalSeconds,
@@ -123,6 +126,17 @@ defaultPostgresPort = 5432
 uniqueDbName :: UniqueTestId -> Text
 uniqueDbName uniqueTestId = "test" <> tshow uniqueTestId
 
+-- * Postgres read replicas
+
+postgresPrimaryPort :: Word16
+postgresPrimaryPort = 65032
+
+postgresReplica1Port :: Word16
+postgresReplica1Port = 65033
+
+postgresReplica2Port :: Word16
+postgresReplica2Port = 65034
+
 -- * Citus
 
 citusPassword :: Text

server/lib/test-harness/src/Harness/GraphqlEngine.hs

@@ -5,6 +5,9 @@
 module Harness.GraphqlEngine
   ( -- * HTTP Calls
 
+    -- ** GET
+    get,
+
     -- ** POST
     post,
     post_,
@@ -56,6 +59,7 @@ import Control.Concurrent (myThreadId)
 import Control.Concurrent.Async qualified as Async
 import Control.Monad.Trans.Managed (ManagedT (..), lowerManagedT)
 import Data.Aeson (Value (String), encode, fromJSON, object, (.=))
+import Data.Aeson qualified as Aeson
 import Data.Aeson.Encode.Pretty as AP
 import Data.Aeson.Types (Pair)
 import Data.ByteString (ByteString)
@@ -77,7 +81,7 @@ import Harness.TestEnvironment (Protocol (..), Server (..), TestEnvironment (..)
 import Harness.WebSockets (responseListener, sendMessages)
 import Hasura.App qualified as App
 import Hasura.Logging (Hasura)
-import Hasura.Prelude
+import Hasura.Prelude hiding (get)
 import Hasura.Server.App (CEConsoleType (OSSConsole))
 import Hasura.Server.Init (PostgresConnInfo (..), ServeOptions (..), unsafePort)
 import Hasura.Server.Metrics (ServerMetricsSpec, createServerMetrics)
@@ -91,6 +95,22 @@ import Test.Hspec
 
 -------------------------------------------------------------------------------
 
+-- HTTP Calls - GET
+
+-- | Get some data from graphql-engine
+--
+-- Optimistically assumes success
+--
+-- Note: We add 'withFrozenCallStack' to reduce stack trace clutter.
+get ::
+  (HasCallStack) => TestEnvironment -> String -> IO Text
+get testEnv@(getServer -> Server {urlPrefix, port}) path =
+  withFrozenCallStack $ do
+    testLogMessage testEnv $ LogHGERequest (T.pack path) $ traceIf testEnv Aeson.Null
+    responseBody <- Http.get (urlPrefix ++ ":" ++ show port ++ path)
+    testLogMessage testEnv $ LogHGEResponse (T.pack path) (String responseBody)
+    pure responseBody
+
 -- HTTP Calls - POST
 
 -- | Post some JSON to graphql-engine, getting back more JSON.

server/lib/test-harness/src/Harness/Http.hs

@@ -3,6 +3,7 @@
 -- | Helper functions for HTTP requests.
 module Harness.Http
   ( get_,
+    get,
     getWithStatus,
     post,
     postValue,
@@ -24,7 +25,7 @@ import Data.Text qualified as T
 import Data.Text.Encoding qualified as T
 import Data.Text.Lazy qualified as TL
 import GHC.Stack
-import Hasura.Prelude
+import Hasura.Prelude hiding (get)
 import Network.HTTP.Client.Conduit qualified as Http.Conduit
 import Network.HTTP.Simple qualified as Http
 import Network.HTTP.Types qualified as Http
@@ -36,16 +37,21 @@ import Text.Pretty.Simple (pShow)
 -- | Performs get, doesn't return the result. Simply throws if there's
 -- not a 200 response.
 get_ :: (HasCallStack) => String -> IO ()
-get_ = getWithStatus [200]
+get_ = void . getWithStatus [200]
+
+-- | Performs get, returns the response body as 'Text'. Throws if there's not a
+-- 200 response.
+get :: (HasCallStack) => String -> IO Text
+get = getWithStatus [200]
 
 -- | Performs get, doesn't return the result. Simply throws if there's
 -- not an expected response status code.
-getWithStatus :: (HasCallStack) => [Int] -> String -> IO ()
+getWithStatus :: (HasCallStack) => [Int] -> String -> IO Text
 getWithStatus acceptableStatusCodes url =
   Http.withResponse @_ @IO (fromString url) \response -> do
     let actualStatusCode = Http.getResponseStatusCode response
+    body <- runConduit $ Http.getResponseBody response .| foldMapC id
     unless (actualStatusCode `elem` acceptableStatusCodes) $ do
-      body <- runConduit $ Http.getResponseBody response .| foldMapC id
       fail
         $ unlines
           [ "The HTTP response had an unexpected response code.",
@@ -55,6 +61,7 @@ getWithStatus acceptableStatusCodes url =
             "Body:",
             T.unpack $ T.decodeUtf8 body
           ]
+    pure $ T.decodeUtf8 body
 
 -- | Post the JSON to the given URL, and produces a very descriptive
 -- exception on failure.

server/src-lib/Hasura/App.hs

@@ -1134,7 +1134,7 @@ mkHGEServer setupHook appStateRef consoleType ekgStore = do
 
   -- initialise the websocket connection reaper thread
   _websocketConnectionReaperThread <-
-    C.forkManagedT "websocket connection reaper thread" logger
+    C.forkManagedT "websocketConnectionReaper" logger
       $ liftIO
       $ WS.websocketConnectionReaper getLatestConfigForWSServer getSchemaCache' (_wseServer wsServerEnv)
 
@@ -1157,7 +1157,7 @@ mkHGEServer setupHook appStateRef consoleType ekgStore = do
   -- set by the user and update the JWK accordingly. This will help in applying the
   -- updates without restarting HGE.
   _ <-
-    C.forkManagedT "update JWK" logger
+    C.forkManagedT "updateJWK" logger
       $ updateJwkCtxThread (getAppContext appStateRef) appEnvManager logger
 
   -- These cleanup actions are not directly associated with any
@@ -1508,7 +1508,8 @@ mkPgSourceResolver pgLogger env sourceName config = runExceptT do
   let context = J.object [("source" J..= sourceName)]
   pgPool <- liftIO $ Q.initPGPool connInfo context connParams pgLogger
   let pgExecCtx = mkPGExecCtx isoLevel pgPool NeverResizePool
-  pure $ PGSourceConfig pgExecCtx connInfo Nothing mempty (pccExtensionsSchema config) mempty ConnTemplate_NotApplicable
+  connInfoWithFinalizer <- liftIO $ mkConnInfoWithFinalizer connInfo (pure ())
+  pure $ PGSourceConfig pgExecCtx connInfoWithFinalizer Nothing mempty (pccExtensionsSchema config) mempty ConnTemplate_NotApplicable
 
 mkMSSQLSourceResolver :: SourceResolver 'MSSQL
 mkMSSQLSourceResolver env _name (MSSQLConnConfiguration connInfo _) = runExceptT do

server/src-lib/Hasura/Backends/Postgres/Execute/Types.hs

@@ -1,5 +1,3 @@
-{-# LANGUAGE QuasiQuotes #-}
-
 -- | Postgres Execute Types
 --
 -- Execution context and source configuration for Postgres databases.
@@ -18,6 +16,8 @@ module Hasura.Backends.Postgres.Execute.Types
 
     -- * Execution in a Postgres Source
     PGSourceConfig (..),
+    getConnInfo,
+    mkConnInfoWithFinalizer,
     ConnectionTemplateConfig (..),
     connectionTemplateConfigResolver,
     ConnectionTemplateResolver (..),
@@ -38,6 +38,8 @@ import Data.Aeson.Extended qualified as J
 import Data.CaseInsensitive qualified as CI
 import Data.Has
 import Data.HashMap.Internal.Strict qualified as Map
+import Data.IORef (IORef)
+import Data.IORef qualified as IORef
 import Data.List.NonEmpty qualified as List.NonEmpty
 import Data.Text.Extended (toTxt)
 import Database.PG.Query qualified as PG
@@ -223,8 +225,8 @@ newtype ConnectionTemplateResolver = ConnectionTemplateResolver
 
 data PGSourceConfig = PGSourceConfig
   { _pscExecCtx :: PGExecCtx,
-    _pscConnInfo :: PG.ConnInfo,
-    _pscReadReplicaConnInfos :: Maybe (NonEmpty PG.ConnInfo),
+    _pscConnInfo :: ConnInfoWithFinalizer,
+    _pscReadReplicaConnInfos :: Maybe (NonEmpty ConnInfoWithFinalizer),
     _pscPostDropHook :: IO (),
     _pscExtensionsSchema :: ExtensionsSchema,
     _pscConnectionSet :: HashMap PostgresConnectionSetMemberName PG.ConnInfo,
@@ -241,11 +243,44 @@ instance Eq PGSourceConfig where
       == (_pscConnInfo rconf, _pscReadReplicaConnInfos rconf, _pscExtensionsSchema rconf, _pscConnectionSet rconf)
 
 instance J.ToJSON PGSourceConfig where
-  toJSON = J.toJSON . show . _pscConnInfo
+  toJSON = J.toJSON . _pscConnInfo
 
 instance Has () PGSourceConfig where
   hasLens = united
 
+-- | Get the primary 'PG.ConnInfo' from 'PGSourceConfig'
+getConnInfo :: PGSourceConfig -> PG.ConnInfo
+getConnInfo = _ciwfConnInfo . _pscConnInfo
+
+-- | Wraps a 'PG.ConnInfo' in a weak IORef with a finalizer action. This is used
+-- to perform any finalizer action (like de-registering metrics), when this
+-- connection info is dropped.
+data ConnInfoWithFinalizer = ConnInfoWithFinalizer
+  { -- | Empty value used to attach a finalizer to (internal)
+    _ciwfWeakIORef :: IORef (),
+    -- | The actual Postgres connection info
+    _ciwfConnInfo :: PG.ConnInfo
+  }
+  deriving (Generic)
+
+instance Show ConnInfoWithFinalizer where
+  show _ = "(ConnInfoWithFinalizer <details>)"
+
+instance Eq ConnInfoWithFinalizer where
+  lconf == rconf = _ciwfConnInfo lconf == _ciwfConnInfo rconf
+
+instance J.ToJSON ConnInfoWithFinalizer where
+  toJSON = J.toJSON . show . _ciwfConnInfo
+
+instance Has () ConnInfoWithFinalizer where
+  hasLens = united
+
+mkConnInfoWithFinalizer :: PG.ConnInfo -> IO () -> IO ConnInfoWithFinalizer
+mkConnInfoWithFinalizer connInfo finalizer = do
+  ioref <- IORef.newIORef ()
+  void $ IORef.mkWeakIORef ioref finalizer
+  pure $ ConnInfoWithFinalizer ioref connInfo
+
 runPgSourceReadTx ::
   (MonadIO m, MonadBaseControl IO m) =>
   PGSourceConfig ->

server/src-lib/Hasura/Server/App.hs

@@ -665,7 +665,7 @@ v1Alpha1PGDumpHandler b = do
       sourceName = PGD.prbSource b
       sourceConfig = unsafeSourceConfiguration @('Postgres 'Vanilla) =<< HashMap.lookup sourceName sources
   ci <-
-    fmap _pscConnInfo sourceConfig
+    fmap getConnInfo sourceConfig
       `onNothing` throw400 NotFound ("source " <> sourceName <<> " not found")
   output <- PGD.execPGDump b ci
   return $ RawResp $ HttpResponse output [sqlHeader]

server/test-postgres/Test/Hasura/EventTriggerCleanupSuite.hs

@@ -44,9 +44,10 @@ buildEventTriggerCleanupSuite = do
   let pgConnInfo = PG.ConnInfo 1 $ PG.CDDatabaseURI $ txtToBs pgUrlText
 
   pgPool <- PG.initPGPool pgConnInfo J.Null PG.defaultConnParams print
+  connInfoWithFinalizer <- liftIO $ mkConnInfoWithFinalizer pgConnInfo (pure ())
 
   let pgContext = mkPGExecCtx PG.ReadCommitted pgPool NeverResizePool
-      dbSourceConfig = PGSourceConfig pgContext pgConnInfo Nothing (pure ()) defaultPostgresExtensionsSchema mempty ConnTemplate_NotApplicable
+      dbSourceConfig = PGSourceConfig pgContext connInfoWithFinalizer Nothing (pure ()) defaultPostgresExtensionsSchema mempty ConnTemplate_NotApplicable
 
   pure $ do
     describe "Event trigger log cleanup" $ eventTriggerLogCleanupSpec dbSourceConfig

server/test-postgres/Test/Hasura/StreamingSubscriptionSuite.hs

@@ -68,9 +68,10 @@ buildStreamingSubscriptionSuite = do
   let pgConnInfo = PG.ConnInfo 1 $ PG.CDDatabaseURI $ txtToBs pgUrlText
 
   pgPool <- PG.initPGPool pgConnInfo J.Null PG.defaultConnParams print
+  connInfoWithFinalizer <- liftIO $ mkConnInfoWithFinalizer pgConnInfo (pure ())
 
   let pgContext = mkPGExecCtx PG.ReadCommitted pgPool NeverResizePool
-      dbSourceConfig = PGSourceConfig pgContext pgConnInfo Nothing (pure ()) defaultPostgresExtensionsSchema mempty ConnTemplate_NotApplicable
+      dbSourceConfig = PGSourceConfig pgContext connInfoWithFinalizer Nothing (pure ()) defaultPostgresExtensionsSchema mempty ConnTemplate_NotApplicable
 
   pure
     $ describe "Streaming subscriptions polling tests"