bump version to 14.8

Fixed integer type mapping in OpenAPI 2.0: replaced the invalid integer format with int32/int64 and added the toSwaggerFormat function to map PostgreSQL types to valid OpenAPI 2.0 formats:

smallint -> int32
integer -> int32
bigint -> int64

.github/workflows/backport.yaml

@@ -45,7 +45,7 @@ jobs:
 
       # Backport action that creates the PR with given settings
       - name: Create backport PR
-        uses: korthout/backport-action@4aaf0e03a94ff0a619c9a511b61aeb42adea5b02 # v4.2.0
+        uses: korthout/backport-action@3c06f323a58619da1e8522229ebc8d5de2633e46 # v4.3.0
         with:
           github_token: ${{ steps.app-token.outputs.token }}
           pull_description: 'Backport for #${pull_number}.'

.github/workflows/release.yaml

@@ -145,7 +145,7 @@ jobs:
         with:
           name: postgrest-ubuntu-aarch64
       - uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd # v4.0.0
-      - uses: docker/login-action@b45d80f862d83dbcd57f89517bcf500b2ab88fb2 # v4.0.0
+      - uses: docker/login-action@4907a6ddec9925e35a0a9e82d7399ccc52663121 # v4.1.0
         with:
           username: ${{ vars.DOCKER_USER }}
           password: ${{ secrets.DOCKER_PASS }}

.github/workflows/test.yaml

@@ -51,7 +51,7 @@ jobs:
       - name: Run coverage (IO tests and Spec tests against PostgreSQL 15)
         run: postgrest-coverage
       - name: Upload coverage to codecov
-        uses: codecov/codecov-action@1af58845a975a7985b0beb0cbe6fbbb71a41dbad # v5.5.3
+        uses: codecov/codecov-action@57e3a136b779b570ffcdbf80b3bdc90e7fab3de2 # v6.0.0
         with:
           files: ./coverage/codecov.json
           token: ${{ secrets.CODECOV_TOKEN }}

CHANGELOG.md

@@ -4,6 +4,16 @@ All notable changes to this project will be documented in this file. From versio
 
 ## Unreleased
 
+## [14.8] - 2026-04-03
+
+### Added
+
+- Log a `HINT` when the LISTEN channel stops working due to a PostgreSQL bug by @laurenceisla in #4581
+
+### Fixed
+
+- Fix invalid OpenAPI 2.0 format for integer types (`smallint`, `integer`, `bigint`) by @arturbent0 in #4641
+
 ## [14.7] - 2026-03-20
 
 ### Fixed

docs/postgrest.dict

@@ -194,3 +194,5 @@ Websockets
 webuser
 wfts
 www
+debouncing
+deduplicates

docs/references/schema_cache.rst

@@ -53,6 +53,19 @@ To reload the schema cache from within the database, you can use the ``NOTIFY``
 
   NOTIFY pgrst, 'reload schema'
 
+Debouncing
+~~~~~~~~~~
+
+PostgREST does not reload the schema cache for each notification when several ``NOTIFY pgrst`` events are generated quickly after one another.
+
+There are two cases to consider: when notifications are sent within a single transaction and when they are sent across multiple transactions.
+
+In the first case, PostgreSQL deduplicates identical ``NOTIFY`` events within the same transaction. This means that even if multiple ``NOTIFY pgrst`` statements are executed before a ``COMMIT``, only a single notification is delivered to PostgREST.
+
+In the second case, when notifications are sent from separate transactions in a short time span, PostgREST applies a debouncing mechanism to avoid excessive schema cache reloads.
+
+Instead of reloading the schema cache for each notification, events are grouped within a small time window of 100 milliseconds. The reload function is executed once immediately when the first notification is received and once more after the burst of events settles, resulting in at most two executions within that time window.
+
 .. _auto_schema_reloading:
 
 Automatic Schema Cache Reloading

postgrest.cabal

@@ -1,5 +1,5 @@
 name:               postgrest
-version:            14.7
+version:            14.8
 synopsis:           REST API for any Postgres database
 description:        Reads the schema of a PostgreSQL database and creates RESTful routes
                     for tables, views, and functions, supporting all HTTP methods that security
@@ -303,6 +303,7 @@ test-suite observability
   main-is:            Main.hs
   other-modules:      ObsHelper
                       Observation.JwtCache
+                      Observation.MetricsSpec
   build-depends:      base              >= 4.9 && < 4.20
                     , base64-bytestring >= 1 && < 1.3
                     , bytestring        >= 0.10.8 && < 0.13
@@ -317,6 +318,7 @@ test-suite observability
                     , postgrest
                     , prometheus-client >= 1.1.1 && < 1.2.0
                     , protolude         >= 0.3.1 && < 0.4
+                    , text              >= 1.2.2 && < 2.2
                     , wai               >= 3.2.1 && < 3.3
   ghc-options:        -threaded -O0 -Werror -Wall -fwarn-identities
                       -fno-spec-constr -optP-Wno-nonportable-include-path

src/PostgREST/Admin.hs

@@ -19,22 +19,23 @@ import PostgREST.Observation (Observation (..))
 
 import qualified PostgREST.AppState as AppState
 
-import Protolude
+import qualified Network.Socket as NS
+import           Protolude
 
-runAdmin :: AppState -> Warp.Settings -> IO ()
-runAdmin appState settings = do
-  whenJust (AppState.getSocketAdmin appState) $ \adminSocket -> do
+runAdmin :: AppState -> Maybe NS.Socket -> NS.Socket -> Warp.Settings -> IO ()
+runAdmin appState maybeAdminSocket socketREST settings = do
+  whenJust maybeAdminSocket $ \adminSocket -> do
     address <- resolveSocketToAddress adminSocket
     observer $ AdminStartObs address
     void . forkIO $ Warp.runSettingsSocket settings adminSocket adminApp
   where
-    adminApp = admin appState
+    adminApp = admin appState socketREST
     observer = AppState.getObserver appState
 
 -- | PostgREST admin application
-admin :: AppState.AppState -> Wai.Application
-admin appState req respond  = do
-  isMainAppReachable  <- isRight <$> reachMainApp (AppState.getSocketREST appState)
+admin :: AppState.AppState -> NS.Socket -> Wai.Application
+admin appState socketREST req respond  = do
+  isMainAppReachable  <- isRight <$> reachMainApp socketREST
   isLoaded <- AppState.isLoaded appState
   isPending <- AppState.isPending appState
 

src/PostgREST/App.hs

@@ -60,10 +60,15 @@ import PostgREST.SchemaCache          (SchemaCache (..))
 import PostgREST.TimeIt               (timeItT)
 import PostgREST.Version              (docsVersion, prettyVersion)
 
-import qualified Data.ByteString.Char8 as BS
-import qualified Data.List             as L
-import qualified Network.HTTP.Types    as HTTP
-import           Protolude             hiding (Handler)
+import qualified Data.ByteString.Char8  as BS
+import qualified Data.List              as L
+import           Data.Streaming.Network (bindPortTCP,
+                                         bindRandomPortTCP)
+import qualified Data.Text              as T
+import qualified Network.HTTP.Types     as HTTP
+import qualified Network.Socket         as NS
+import           PostgREST.Unix         (createAndBindDomainSocket)
+import           Protolude              hiding (Handler)
 
 type Handler = ExceptT Error
 
@@ -72,19 +77,21 @@ run appState = do
   conf@AppConfig{..} <- AppState.getConfig appState
 
   AppState.schemaCacheLoader appState -- Loads the initial SchemaCache
+  (mainSocket, adminSocket) <- initSockets conf
+
   Unix.installSignalHandlers observer (AppState.getMainThreadId appState) (AppState.schemaCacheLoader appState) (AppState.readInDbConfig False appState)
 
   Listener.runListener appState
 
-  Admin.runAdmin appState (serverSettings conf)
+  Admin.runAdmin appState adminSocket mainSocket (serverSettings conf)
 
   let app = postgrest configLogLevel appState (AppState.schemaCacheLoader appState)
 
   do
-    address <- resolveSocketToAddress (AppState.getSocketREST appState)
+    address <- resolveSocketToAddress mainSocket
     observer $ AppServerAddressObs address
 
-  Warp.runSettingsSocket (serverSettings conf & setOnException onWarpException) (AppState.getSocketREST appState) app
+  Warp.runSettingsSocket (serverSettings conf & setOnException onWarpException) mainSocket app
   where
     observer = AppState.getObserver appState
 
@@ -228,3 +235,40 @@ addRetryHint delay response = do
 
 isServiceUnavailable :: Wai.Response -> Bool
 isServiceUnavailable response = Wai.responseStatus response == HTTP.status503
+
+type AppSockets = (NS.Socket, Maybe NS.Socket)
+
+initSockets :: AppConfig -> IO AppSockets
+initSockets AppConfig{..} = do
+  let
+    cfg'usp = configServerUnixSocket
+    cfg'uspm = configServerUnixSocketMode
+    cfg'host = configServerHost
+    cfg'port = configServerPort
+    cfg'adminHost = configAdminServerHost
+    cfg'adminPort = configAdminServerPort
+
+  sock <- case cfg'usp of
+    -- I'm not using `streaming-commons`' bindPath function here because it's not defined for Windows,
+    -- but we need to have runtime error if we try to use it in Windows, not compile time error
+    Just path -> createAndBindDomainSocket path cfg'uspm
+    Nothing -> do
+      (_, sock) <-
+        if cfg'port /= 0
+          then do
+            sock <- bindPortTCP cfg'port (fromString $ T.unpack cfg'host)
+            pure (cfg'port, sock)
+          else do
+            -- explicitly bind to a random port, returning bound port number
+            (num, sock) <- bindRandomPortTCP (fromString $ T.unpack cfg'host)
+            pure (num, sock)
+      pure sock
+
+  adminSock <- case cfg'adminPort of
+    Just adminPort -> do
+      adminSock <- bindPortTCP adminPort (fromString $ T.unpack cfg'adminHost)
+      pure $ Just adminSock
+    Nothing -> pure Nothing
+
+  pure (sock, adminSock)
+

src/PostgREST/AppState.hs

@@ -13,11 +13,9 @@ module PostgREST.AppState
   , getNextListenerDelay
   , getTime
   , getJwtCacheState
-  , getSocketREST
-  , getSocketAdmin
   , init
-  , initSockets
   , initWithPool
+  , putConfig -- For tests TODO refactoring
   , putNextListenerDelay
   , putSchemaCache
   , putPgVersion
@@ -32,13 +30,11 @@ module PostgREST.AppState
 
 import qualified Data.ByteString.Char8      as BS
 import           Data.Either.Combinators    (whenLeft)
-import qualified Data.Text                  as T (unpack)
 import qualified Hasql.Pool                 as SQL
 import qualified Hasql.Pool.Config          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.Socket             as NS
 import qualified PostgREST.Auth.JwtCache    as JwtCache
 import qualified PostgREST.Error            as Error
 import qualified PostgREST.Logger           as Logger
@@ -70,10 +66,7 @@ import PostgREST.SchemaCache             (SchemaCache (..),
                                           querySchemaCache,
                                           showSummary)
 import PostgREST.SchemaCache.Identifiers (quoteQi)
-import PostgREST.Unix                    (createAndBindDomainSocket)
 
-import Data.Streaming.Network (bindPortTCP, bindRandomPortTCP)
-import Data.String            (IsString (..))
 import Protolude
 
 data AppState = AppState
@@ -99,10 +92,6 @@ data AppState = AppState
   , stateNextDelay         :: IORef Int
   -- | Keeps track of the next delay for the listener
   , stateNextListenerDelay :: IORef Int
-  -- | Network socket for REST API
-  , stateSocketREST        :: NS.Socket
-  -- | Network socket for the admin UI
-  , stateSocketAdmin       :: Maybe NS.Socket
   -- | Observation handler
   , stateObserver          :: ObservationHandler
   -- | JWT Cache
@@ -117,8 +106,6 @@ data SchemaCacheStatus
   | SCPending
   deriving Eq
 
-type AppSockets = (NS.Socket, Maybe NS.Socket)
-
 init :: AppConfig -> IO AppState
 init conf@AppConfig{configLogLevel, configDbPoolSize} = do
   loggerState  <- Logger.init
@@ -128,12 +115,10 @@ init conf@AppConfig{configLogLevel, configDbPoolSize} = do
   observer $ AppStartObs prettyVersion
 
   pool <- initPool conf observer
-  (sock, adminSock) <- initSockets conf
-  state' <- initWithPool (sock, adminSock) pool conf loggerState metricsState observer
-  pure state' { stateSocketREST = sock, stateSocketAdmin = adminSock}
+  initWithPool pool conf loggerState metricsState observer --{ stateSocketREST = sock, stateSocketAdmin = adminSock}
 
-initWithPool :: AppSockets -> SQL.Pool -> AppConfig -> Logger.LoggerState -> Metrics.MetricsState -> ObservationHandler -> IO AppState
-initWithPool (sock, adminSock) pool conf loggerState metricsState observer = do
+initWithPool :: SQL.Pool -> AppConfig -> Logger.LoggerState -> Metrics.MetricsState -> ObservationHandler -> IO AppState
+initWithPool pool conf loggerState metricsState observer = do
 
   appState <- AppState pool
     <$> newIORef minimumPgVersion -- assume we're in a supported version when starting, this will be corrected on a later step
@@ -146,8 +131,6 @@ initWithPool (sock, adminSock) pool conf loggerState metricsState observer = do
     <*> myThreadId
     <*> newIORef 0
     <*> newIORef 1
-    <*> pure sock
-    <*> pure adminSock
     <*> pure observer
     <*> JwtCache.init conf observer
     <*> pure loggerState
@@ -166,40 +149,6 @@ initWithPool (sock, adminSock) pool conf loggerState metricsState observer = do
 destroy :: AppState -> IO ()
 destroy = destroyPool
 
-initSockets :: AppConfig -> IO AppSockets
-initSockets AppConfig{..} = do
-  let
-    cfg'usp = configServerUnixSocket
-    cfg'uspm = configServerUnixSocketMode
-    cfg'host = configServerHost
-    cfg'port = configServerPort
-    cfg'adminHost = configAdminServerHost
-    cfg'adminPort = configAdminServerPort
-
-  sock <- case cfg'usp of
-    -- I'm not using `streaming-commons`' bindPath function here because it's not defined for Windows,
-    -- but we need to have runtime error if we try to use it in Windows, not compile time error
-    Just path -> createAndBindDomainSocket path cfg'uspm
-    Nothing -> do
-      (_, sock) <-
-        if cfg'port /= 0
-          then do
-            sock <- bindPortTCP cfg'port (fromString $ T.unpack cfg'host)
-            pure (cfg'port, sock)
-          else do
-            -- explicitly bind to a random port, returning bound port number
-            (num, sock) <- bindRandomPortTCP (fromString $ T.unpack cfg'host)
-            pure (num, sock)
-      pure sock
-
-  adminSock <- case cfg'adminPort of
-    Just adminPort -> do
-      adminSock <- bindPortTCP adminPort (fromString $ T.unpack cfg'adminHost)
-      pure $ Just adminSock
-    Nothing -> pure Nothing
-
-  pure (sock, adminSock)
-
 initPool :: AppConfig -> ObservationHandler -> IO SQL.Pool
 initPool AppConfig{..} observer = do
   SQL.acquire $ SQL.settings
@@ -313,12 +262,6 @@ getTime = stateGetTime
 getJwtCacheState :: AppState -> JwtCacheState
 getJwtCacheState = stateJwtCache
 
-getSocketREST :: AppState -> NS.Socket
-getSocketREST = stateSocketREST
-
-getSocketAdmin :: AppState -> Maybe NS.Socket
-getSocketAdmin = stateSocketAdmin
-
 getMainThreadId :: AppState -> ThreadId
 getMainThreadId = stateMainThreadId
 

src/PostgREST/Listener.hs

@@ -10,7 +10,8 @@ import qualified Hasql.Connection      as SQL
 import qualified Hasql.Notifications   as SQL
 import           PostgREST.AppState    (AppState, getConfig)
 import           PostgREST.Config      (AppConfig (..))
-import           PostgREST.Observation (Observation (..))
+import           PostgREST.Observation (Observation (..),
+                                        isDbListenerBug)
 import           PostgREST.Version     (prettyVersion)
 
 import qualified PostgREST.AppState as AppState
@@ -36,6 +37,8 @@ retryingListen appState = do
     onError err = do
       AppState.putIsListenerOn appState False
       observer $ DBListenFail dbChannel (Right err)
+      when (isDbListenerBug err) $
+        observer DBListenBugHint
       unless configDbPoolAutomaticRecovery $
         killThread mainThreadId
 

src/PostgREST/Observation.hs

@@ -1,4 +1,5 @@
-{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE DeriveGeneric #-}
+{-# LANGUAGE LambdaCase    #-}
 {-|
 Module      : PostgREST.Observation
 Description : This module holds an Observation type which is the core of Observability for PostgREST.
@@ -12,6 +13,7 @@ module PostgREST.Observation
   , observationMessage
   , ObservationHandler
   , showOnSingleLine
+  , isDbListenerBug
   ) where
 
 import qualified Data.ByteString.Lazy       as LBS
@@ -46,6 +48,7 @@ data Observation
   | DBListenStart Text
   | DBListenFail Text (Either SQL.ConnectionError SomeException)
   | DBListenRetry Int
+  | DBListenBugHint -- https://github.com/PostgREST/postgrest/issues/3147
   | DBListenerGotSCacheMsg ByteString
   | DBListenerGotConfigMsg ByteString
   | DBListenerConnectionCleanupFail SomeException
@@ -65,6 +68,7 @@ data Observation
   | JwtCacheEviction
   | TerminationUnixSignalObs Text
   | WarpErrorObs Text
+  deriving (Generic)
 
 data ObsFatalError = ServerAuthError | ServerPgrstBug | ServerError42P05 | ServerError08P01
 
@@ -117,6 +121,8 @@ observationMessage = \case
       either showListenerConnError showListenerException listenErr
   DBListenRetry delay ->
     "Retrying listening for database notifications in " <> (show delay::Text) <> " seconds..."
+  DBListenBugHint ->
+    "HINT:  This is likely a bug in the notification queue, try executing the following to solve it: select pg_notification_queue_usage();"
   DBListenerGotSCacheMsg channel ->
     "Received a schema cache reload message on the " <> show channel <> " channel"
   DBListenerGotConfigMsg channel ->
@@ -179,3 +185,6 @@ observationMessage = \case
 
 showOnSingleLine :: Char -> Text -> Text
 showOnSingleLine split txt = T.intercalate " " $ T.filter (/= split) <$> T.lines txt -- the errors from hasql-notifications come intercalated with "\t\n"
+
+isDbListenerBug :: SomeException -> Bool
+isDbListenerBug e = "could not access status of transaction" `T.isInfixOf` show e

src/PostgREST/Response/OpenAPI.hs

@@ -73,6 +73,12 @@ toSwaggerType colType             = case T.takeEnd 2 colType of
   "[]" -> Just SwaggerArray
   _    -> Just SwaggerString
 
+toSwaggerFormat :: Text -> Maybe Text
+toSwaggerFormat "smallint" = Just "int32"
+toSwaggerFormat "integer"  = Just "int32"
+toSwaggerFormat "bigint"   = Just "int64"
+toSwaggerFormat colType    = Just colType
+
 typeFromArray :: Text -> Text
 typeFromArray = T.dropEnd 2
 
@@ -141,7 +147,7 @@ makeProperty tbl rels col = (colName col, Inline s)
         & default_ .~ (JSON.decode . toUtf8Lazy . parseDefault (colType col) =<< colDefault col)
         & description .~ d
         & enum_ .~ e
-        & format ?~ colType col
+        & format .~ toSwaggerFormat (colType col)
         & maxLength .~ (fromIntegral <$> colMaxLen col)
         & type_ .~ toSwaggerType (colType col)
         & items .~ (SwaggerItemsObject <$> makePropertyItems (colType col))
@@ -160,7 +166,7 @@ makeProcProperty (RoutineParam n t _ _ _) = (n, Inline s)
     s = (mempty :: Schema)
           & type_ .~ toSwaggerType t
           & items .~ (SwaggerItemsObject <$> makePropertyItems t)
-          & format ?~ t
+          & format .~ toSwaggerFormat t
 
 makePreferParam :: [Text] -> Param
 makePreferParam ts =
@@ -192,14 +198,14 @@ makeProcGetParam (RoutineParam n t _ r v) =
     baseSchema = (mempty :: ParamOtherSchema)
       & in_ .~ ParamQuery
     schemaNotMulti = baseSchema
-      & format ?~ t
+      & format .~ toSwaggerFormat t
       & type_ ?~ toParamType (toSwaggerType t)
     schemaMulti = baseSchema
       & type_ ?~ fromMaybe SwaggerString (toSwaggerType t)
       & items ?~ SwaggerItemsPrimitive (Just CollectionMulti)
         ((mempty :: ParamSchema x)
           & type_ .~ toSwaggerTypeFromArray t
-          & format ?~ typeFromArray t)
+          & format .~ toSwaggerFormat (typeFromArray t))
     toParamType paramType = case paramType of
       -- Array uses {} in query params
       Just SwaggerArray -> SwaggerString

test/io/test_big_schema.py

@@ -109,4 +109,4 @@ def test_second_request_for_non_existent_table_should_be_quick(defaultenv):
         assert data["code"] == "PGRST205"
         first_duration = response.elapsed.total_seconds()
         response = postgrest.session.get("/unknown-table")
-        assert response.elapsed.total_seconds() < first_duration / 10
+        assert response.elapsed.total_seconds() < first_duration / 2

test/observability/Main.hs

@@ -15,40 +15,55 @@ import qualified PostgREST.Metrics         as Metrics
 import           PostgREST.SchemaCache     (querySchemaCache)
 
 import qualified Observation.JwtCache
+import qualified Observation.MetricsSpec
 
 import ObsHelper
-import Protolude  hiding (toList, toS)
+import PostgREST.Observation (Observation (HasqlPoolObs))
+import Protolude             hiding (toList, toS)
 import Test.Hspec
 
 main :: IO ()
 main = do
+  poolChan <- newChan
+  -- make sure poolChan is not growing indefinitely
+  -- start a thread that drains the channel
+  -- this is necessary because test cases operate on
+  -- copies so poolChan is never read from
+  -- this means we have another thread running for the entire duration of the spec but this shouldn't be a problem since Haskell green threads are lightweight
+  void $ forkIO $ forever $ readChan poolChan
+  metricsState <- Metrics.init (configDbPoolSize testCfg)
   pool <- P.acquire $ P.settings
     [ P.size 3
     , P.acquisitionTimeout 10
     , P.agingTimeout 60
     , P.idlenessTimeout 60
     , P.staticConnectionSettings (toUtf8 $ configDbUri testCfg)
+    -- make sure metrics are updated and pool observations published to poolChan
+    , P.observationHandler $ (writeChan poolChan <> Metrics.observationMetrics metricsState) . HasqlPoolObs
     ]
 
   actualPgVersion <- either (panic . show) id <$> P.use pool (queryPgVersion False)
 
   -- cached schema cache so most tests run fast
   baseSchemaCache <- loadSCache pool testCfg
-  sockets <- AppState.initSockets testCfg
   loggerState <- Logger.init
-  metricsState <- Metrics.init (configDbPoolSize testCfg)
 
   let
-    initApp sCache st config = do
-      appState <- AppState.initWithPool sockets pool config loggerState metricsState (Metrics.observationMetrics metricsState)
+    initApp sCache config = do
+      -- duplicate poolChan as a starting point
+      obsChan <- dupChan poolChan
+      stateObsChan <- newObsChan obsChan
+      appState <- AppState.initWithPool pool config loggerState metricsState (Metrics.observationMetrics metricsState <> writeChan obsChan)
       AppState.putPgVersion appState actualPgVersion
       AppState.putSchemaCache appState (Just sCache)
-      return (st, postgrest (configLogLevel config) appState (pure ()))
+      return (SpecState appState metricsState stateObsChan, postgrest (configLogLevel config) appState (pure ()))
 
   -- Run all test modules
   hspec $ do
-    before (initApp baseSchemaCache metricsState testCfgJwtCache) $
+    before (initApp baseSchemaCache testCfgJwtCache) $
       describe "Observation.JwtCacheObs" Observation.JwtCache.spec
+    before (initApp baseSchemaCache testCfg) $
+      describe "Feature.MetricsSpec" Observation.MetricsSpec.spec
 
   where
     loadSCache pool conf =

test/observability/ObsHelper.hs

@@ -1,31 +1,69 @@
 {-# LANGUAGE AllowAmbiguousTypes       #-}
+{-# LANGUAGE DeriveAnyClass            #-}
 {-# LANGUAGE ExistentialQuantification #-}
 {-# LANGUAGE FlexibleContexts          #-}
+{-# LANGUAGE FlexibleInstances         #-}
+{-# LANGUAGE LambdaCase                #-}
+{-# LANGUAGE RankNTypes                #-}
 {-# LANGUAGE ScopedTypeVariables       #-}
 {-# LANGUAGE TupleSections             #-}
 {-# LANGUAGE TypeApplications          #-}
+{-# LANGUAGE TypeOperators             #-}
 module ObsHelper where
 
-import qualified Data.ByteString.Base64 as B64 (decodeLenient)
-import qualified Data.ByteString.Char8  as BS
-import qualified Data.ByteString.Lazy   as BL
-import qualified Jose.Jwa               as JWT
-import qualified Jose.Jws               as JWT
-import qualified Jose.Jwt               as JWT
+import qualified Data.ByteString                 as BS
+import qualified Data.ByteString.Base64          as B64
+import qualified Data.ByteString.Lazy            as BL
+import qualified Data.List                       as DL
+import           Data.List.NonEmpty              (fromList)
+import           Data.String                     (String)
+import qualified Data.Text                       as T
+import qualified Jose.Jwa                        as JWT
+import qualified Jose.Jws                        as JWT
+import qualified Jose.Jwt                        as JWT
+import           Network.HTTP.Types
+import qualified PostgREST.AppState              as AppState
+import           PostgREST.Config                (AppConfig (..),
+                                                  JSPathExp (..),
+                                                  LogLevel (..),
+                                                  OpenAPIMode (..),
+                                                  parseSecret)
+import qualified PostgREST.Metrics               as Metrics
+import           PostgREST.Observation           (Observation (..))
+import           Prometheus                      (Counter, getCounter)
+import           Protolude                       hiding (get, toS)
+import           System.Timeout                  (timeout)
+import           Test.Hspec
+import           Test.Hspec.Expectations.Contrib (annotate)
+-- helpers used to produce observation diagnostics in waitForObs
+-- Implementing the Show instance for Observation is hard due to having many different parameters so instead we use generic programming (`conName`) to obtain the constructor name as `Text`
+class HasConstructor f where
+  genericConstrName :: f x -> Text
 
-import PostgREST.Config (AppConfig (..), JSPathExp (..),
-                         LogLevel (..), OpenAPIMode (..), parseSecret)
+instance HasConstructor f => HasConstructor (D1 c f) where
+  genericConstrName (M1 x) = genericConstrName x
 
-import Data.List.NonEmpty              (fromList)
-import Data.String                     (String)
-import Prometheus                      (Counter, getCounter)
-import Test.Hspec.Expectations.Contrib (annotate)
+instance (HasConstructor x, HasConstructor y) => HasConstructor (x :+: y) where
+  genericConstrName (L1 l) = genericConstrName l
+  genericConstrName (R1 r) = genericConstrName r
 
-import Network.HTTP.Types
-import Protolude
-import Test.Hspec
-import Test.Hspec.Wai
+instance Constructor c => HasConstructor (C1 c f) where
+  genericConstrName = T.pack . conName
 
+data SpecState = SpecState {
+  specAppState :: AppState.AppState,
+  specMetrics  :: Metrics.MetricsState,
+  specObsChan  :: ObsChan
+}
+
+data StateCheck st m = forall a. StateCheck (st -> (String, m a)) (a -> a -> Expectation)
+
+data TimeoutException = TimeoutException deriving (Show, Exception)
+
+data ObsChan = ObsChan (Chan Observation) (Chan Observation)
+
+constrName :: (HasConstructor (Rep a), Generic a)=> a -> Text
+constrName = genericConstrName . from
 
 baseCfg :: AppConfig
 baseCfg = let secret = encodeUtf8 "reallyreallyreallyreallyverysafe" in
@@ -107,18 +145,12 @@ generateJWT claims =
   either mempty JWT.unJwt $ JWT.hmacEncode JWT.HS256 generateSecret (BL.toStrict claims)
 
 -- state check helpers
-
-data StateCheck st m = forall a. StateCheck (st -> (String, m a)) (a -> a -> Expectation)
-
 stateCheck :: (Show a, Eq a) => (c -> m a) -> (st -> (String, c)) -> (a -> a) -> StateCheck st m
 stateCheck extractValue extractComponent expect = StateCheck (second extractValue . extractComponent) (flip shouldBe . expect)
 
 expectField :: forall s st a c m. (KnownSymbol s, Show a, Eq a, HasField s st c) => (c -> m a) -> (a -> a) -> StateCheck st m
 expectField extractValue = stateCheck extractValue ((symbolVal (Proxy @s),) . getField @s)
 
-checkState :: (Traversable t) => t (StateCheck st (WaiSession st)) -> WaiSession st b -> WaiSession st ()
-checkState checks act = getState >>= flip (`checkState'` checks) act
-
 checkState' :: (Traversable t, MonadIO m) => st -> t (StateCheck st m) -> m b -> m ()
 checkState' initialState checks act = do
   expectations <- traverse (\(StateCheck g expect) -> let (msg, m) = g initialState in m >>= createExpectation msg m . expect) checks
@@ -131,3 +163,48 @@ expectCounter :: forall s st m. (KnownSymbol s, HasField s st Counter, MonadIO m
 expectCounter = expectField @s intCounter
   where
     intCounter = ((round @Double @Int) <$>) . getCounter
+
+accumulateUntilTimeout :: Int -> (s -> a -> s) -> s -> IO a -> IO s
+accumulateUntilTimeout t f start act = do
+  tid <- myThreadId
+  -- mask to make sure TimeoutException is not thrown before starting the loop
+  mask $ \unmask -> do
+    -- start timeout thread unmasking exceptions
+    ttid <- forkIOWithUnmask ($ (threadDelay t *> throwTo tid TimeoutException))
+    -- unmask effect
+    unmask (fix (\loop accum -> (act >>= loop . f accum) `onTimeout` pure accum) start)
+      -- make sure we catch timeout if happens before entering the loop
+      `onTimeout` pure start
+      -- make sure timer thread is killed on other exceptions
+      -- so that it won't throw TimeoutException later
+      `onException` killThread ttid
+  where
+    onTimeout m a = m `catch` \TimeoutException -> a
+
+newObsChan :: Chan Observation -> IO ObsChan
+newObsChan = fmap <$> ObsChan <*> dupChan
+
+-- read messages from copy chan and once condition is met drain original to the same point
+-- upon timeout report error and messages remaining in the original chan
+-- that way we report messages since last successful read
+waitForObs :: HasCallStack => ObsChan -> Int -> Text -> (Observation -> Maybe a) -> IO ()
+waitForObs (ObsChan orig copy) t msg f =
+  timeout t (readUntil copy *> readUntil orig) >>= maybe failTimeout mempty
+  where
+    failTimeout = takeUntilTimeout decisecond (readChan orig)
+        >>= expectationFailure . DL.unlines . fmap show . (failureMessageHeader :) . fmap obsDiagMessage
+    failureMessageHeader = "Timeout waiting for " <> msg <> " at " <> loc <> ". Remaining observations:"
+    readUntil = void . untilM (pure . not . null . f) . readChan
+    loc = fromMaybe "(unknown)" . head $ (T.pack . prettySrcLoc . snd <$> getCallStack callStack)
+    -- execute effectful computation until result meets provided condition
+    untilM cond m = fix $ \loop -> m >>= \a -> ifM (cond a) (pure a) loop
+    -- duplicate the provided channel and construct wairFor function binding both channels
+    -- accumulate effecful computation results into a list for specified time
+    takeUntilTimeout t' = fmap reverse . accumulateUntilTimeout t' (flip (:)) []
+    decisecond = 100000
+
+obsDiagMessage :: Observation -> Text
+obsDiagMessage = \case
+  (HasqlPoolObs o) -> show o
+  o@(DBListenStart channel) -> constrName o <> show channel
+  o -> constrName o

test/observability/Observation/JwtCache.hs

@@ -1,4 +1,5 @@
 {-# LANGUAGE DataKinds        #-}
+{-# LANGUAGE NamedFieldPuns   #-}
 {-# LANGUAGE TypeApplications #-}
 module Observation.JwtCache where
 
@@ -13,9 +14,11 @@ import PostgREST.Metrics   (MetricsState (..))
 import Protolude
 import Test.Hspec.Wai.JSON (json)
 
-spec :: SpecWith (MetricsState, Application)
+spec :: SpecWith (SpecState, Application)
 spec = describe "Server started with JWT and metrics enabled" $ do
   it "Should not have JWT in cache" $ do
+    expectCounters <- checkState' . specMetrics <$> getState
+
     let auth = genToken [json|{"exp": 9999999999, "role": "postgrest_test_author", "id": "jdoe1"}|]
 
     expectCounters
@@ -27,6 +30,8 @@ spec = describe "Server started with JWT and metrics enabled" $ do
          request methodGet "/authors_only" [auth] "" `shouldRespondWith` 200
 
   it "Should have JWT in cache" $ do
+    expectCounters <- checkState' . specMetrics <$> getState
+
     let auth = genToken [json|{"exp": 9999999999, "role": "postgrest_test_author", "id": "jdoe2"}|]
 
     expectCounters
@@ -39,6 +44,8 @@ spec = describe "Server started with JWT and metrics enabled" $ do
       *> request methodGet "/authors_only" [auth] "" `shouldRespondWith` 200
 
   it "Should not cache invalid JWTs" $ do
+    expectCounters <- checkState' . specMetrics <$> getState
+
     let auth = authHeaderJWT "some random bytes"
 
     expectCounters
@@ -51,6 +58,8 @@ spec = describe "Server started with JWT and metrics enabled" $ do
       *> request methodGet "/authors_only" [auth] "" `shouldRespondWith` 401
 
   it "Should cache expired JWTs" $ do
+    expectCounters <- checkState' . specMetrics <$> getState
+
     let auth = genToken [json|{"exp": 1, "role": "postgrest_test_author", "id": "jdoe2"}|]
 
     expectCounters
@@ -63,6 +72,8 @@ spec = describe "Server started with JWT and metrics enabled" $ do
       *> request methodGet "/authors_only" [auth] "" `shouldRespondWith` 401
 
   it "Should evict entries from the JWT cache (jwt cache max is 2)" $ do
+    expectCounters <- checkState' . specMetrics <$> getState
+
     let jwt1 = genToken [json|{"exp": 9999999999, "role": "postgrest_test_author", "id": "jdoe3"}|]
         jwt2 = genToken [json|{"exp": 9999999999, "role": "postgrest_test_author", "id": "jdoe4"}|]
         jwt3 = genToken [json|{"exp": 9999999999, "role": "postgrest_test_author", "id": "jdoe5"}|]
@@ -82,6 +93,8 @@ spec = describe "Server started with JWT and metrics enabled" $ do
       *> request methodGet "/authors_only" [jwt3] ""
 
   it "Should not evict entries from the JWT cache in FIFO order" $ do
+    expectCounters <- checkState' . specMetrics <$> getState
+
     let jwt1 = genToken [json|{"exp": 9999999999, "role": "postgrest_test_author", "id": "jdoe6"}|]
         jwt2 = genToken [json|{"exp": 9999999999, "role": "postgrest_test_author", "id": "jdoe7"}|]
         jwt3 = genToken [json|{"exp": 9999999999, "role": "postgrest_test_author", "id": "jdoe8"}|]
@@ -108,6 +121,8 @@ spec = describe "Server started with JWT and metrics enabled" $ do
   -- The test case was added based on coverage report
   -- showing this scenario was not covered by previous tests
   it "Should evict entries even though all were hit" $ do
+    expectCounters <- checkState' . specMetrics <$> getState
+
     let jwt1 = genToken [json|{"exp": 9999999999, "role": "postgrest_test_author", "id": "jdoe9"}|]
         jwt2 = genToken [json|{"exp": 9999999999, "role": "postgrest_test_author", "id": "jdoe10"}|]
         jwt3 = genToken [json|{"exp": 9999999999, "role": "postgrest_test_author", "id": "jdoe11"}|]
@@ -135,4 +150,3 @@ spec = describe "Server started with JWT and metrics enabled" $ do
       requests = expectCounter @"jwtCacheRequests"
       hits = expectCounter @"jwtCacheHits"
       evictions = expectCounter @"jwtCacheEvictions"
-      expectCounters = checkState

test/observability/Observation/MetricsSpec.hs

@@ -0,0 +1,79 @@
+{-# LANGUAGE DataKinds           #-}
+{-# LANGUAGE FlexibleContexts    #-}
+{-# LANGUAGE MonadComprehensions #-}
+{-# LANGUAGE NamedFieldPuns      #-}
+{-# LANGUAGE TypeApplications    #-}
+
+module Observation.MetricsSpec where
+
+import           Data.List             (lookup)
+import           Network.Wai           (Application)
+import           ObsHelper
+import qualified PostgREST.AppState    as AppState
+import           PostgREST.Config      (AppConfig (configDbSchemas))
+import qualified PostgREST.Metrics     as Metrics
+import           PostgREST.Observation
+import           Prometheus            (getCounter, getVectorWith)
+import           Protolude
+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 update pgrst_schema_cache_loads_total[SUCCESS]" $ do
+    SpecState{specAppState = appState, specMetrics = metrics, specObsChan} <- getState
+    let waitFor = waitForObs specObsChan
+
+    liftIO $ checkState' metrics [
+        schemaCacheLoads "SUCCESS" (+1)
+      ] $ do
+        AppState.schemaCacheLoader appState
+        waitFor (1 * sec) "SchemaCacheLoadedObs" $ \x -> [ o | o@(SchemaCacheLoadedObs{}) <- pure x]
+
+  it "Should update pgrst_schema_cache_loads_total[ERROR]" $ do
+    SpecState{specAppState = appState, specMetrics = metrics, specObsChan} <- getState
+    let waitFor = waitForObs specObsChan
+
+    liftIO $ checkState' metrics [
+        schemaCacheLoads "FAIL" (+1),
+        schemaCacheLoads "SUCCESS" (+1)
+      ] $ do
+        AppState.getConfig appState >>= \prev -> do
+          AppState.putConfig appState $ prev { configDbSchemas = pure "bad_schema" }
+          AppState.schemaCacheLoader appState
+          waitFor (1 * sec) "SchemaCacheErrorObs" $ \x -> [ o | o@(SchemaCacheErrorObs{}) <- pure x]
+          AppState.putConfig appState prev
+
+        -- wait up to 2 secs so that retry can happen
+        waitFor (2 * sec) "SchemaCacheLoadedObs" $ \x -> [ o | o@(SchemaCacheLoadedObs{}) <- pure x]
+
+  it "Should debounce schema cache loads" $ do
+    SpecState{specAppState = appState, specMetrics = metrics, specObsChan} <- getState
+    let waitFor = waitForObs specObsChan
+
+    liftIO $ checkState' metrics [
+        -- we expect exactly 2 successful schema cache loads
+        schemaCacheLoads "FAIL" (+0),
+        schemaCacheLoads "SUCCESS" (+2)
+      ] $ do
+        AppState.schemaCacheLoader appState
+        -- at this moment there is no dedicated observation emited
+        -- when schema cache load starts
+        -- so we wait for DBConnectedObs which is emited right after successful
+        -- PostgreSQL version query during schema cache loading
+        waitFor (1 * sec) "DBConnectedObs" $ \x -> [ o | o@(DBConnectedObs{}) <- pure x]
+        -- request schema cache load multiple times
+        -- all of them should be handled by a single schema cache load
+        replicateM_ 100 (AppState.schemaCacheLoader appState)
+        -- wait for two expected SchemaCacheLoadedObs events
+        replicateM_ 2 $ waitFor (1 * sec) "SchemaCacheLoadedObs" $ \x -> [ o | o@(SchemaCacheLoadedObs{}) <- pure x]
+        -- wait 1 sec to make sure we capture all potential schema cache loads
+        -- (there should be none but we need to verify that)
+        threadDelay $ 1 * sec
+
+
+  where
+    -- prometheus-client api to handle vectors is convoluted
+    schemaCacheLoads label = expectField @"schemaCacheLoads" $
+      fmap (maybe (0::Int) round . lookup label) . (`getVectorWith` getCounter)
+    sec = 1000000

test/spec/Feature/OpenApi/OpenApiSpec.hs

@@ -157,7 +157,7 @@ spec = describe "OpenAPI" $ do
               "properties": {
                 "id": {
                   "description": "child_entities id comment\n\nNote:\nThis is a Primary Key.<pk/>",
-                  "format": "integer",
+                  "format": "int32",
                   "type": "integer"
                 },
                 "name": {
@@ -167,7 +167,7 @@ spec = describe "OpenAPI" $ do
                 },
                 "parent_id": {
                   "description": "Note:\nThis is a Foreign Key to `entities.id`.<fk table='entities' column='id'/>",
-                  "format": "integer",
+                  "format": "int32",
                   "type": "integer"
                 }
               },
@@ -192,7 +192,7 @@ spec = describe "OpenAPI" $ do
               "properties": {
                 "id": {
                   "description": "child_entities_view id comment\n\nNote:\nThis is a Primary Key.<pk/>",
-                  "format": "integer",
+                  "format": "int32",
                   "type": "integer"
                 },
                 "name": {
@@ -202,7 +202,7 @@ spec = describe "OpenAPI" $ do
                 },
                 "parent_id": {
                   "description": "Note:\nThis is a Foreign Key to `entities.id`.<fk table='entities' column='id'/>",
-                  "format": "integer",
+                  "format": "int32",
                   "type": "integer"
                 }
               }
@@ -232,7 +232,7 @@ spec = describe "OpenAPI" $ do
         referralLink `shouldBe` Just
           [aesonQQ|
             {
-              "format": "integer",
+              "format": "int32",
               "type": "integer",
               "description": "Note:\nThis is a Foreign Key to `second.id`.<fk table='second' column='id'/>"
             }
@@ -337,7 +337,7 @@ spec = describe "OpenAPI" $ do
         referralLink `shouldBe` Just
           [aesonQQ|
             {
-              "format": "integer",
+              "format": "int32",
               "type": "integer",
               "description": "Note:\nThis is a Foreign Key to `pages.link`.<fk table='pages' column='link'/>"
             }
@@ -354,7 +354,7 @@ spec = describe "OpenAPI" $ do
         referralLink `shouldBe` Just
           [aesonQQ|
             {
-              "format": "integer",
+              "format": "int32",
               "type": "integer",
               "description": "Note:\nThis is a Foreign Key to `clients.id`.<fk table='clients' column='id'/>"
             }
@@ -432,7 +432,7 @@ spec = describe "OpenAPI" $ do
         types `shouldBe` Just
           [aesonQQ|
             {
-              "format": "smallint",
+              "format": "int32",
               "type": "integer"
             }
           |]
@@ -447,7 +447,7 @@ spec = describe "OpenAPI" $ do
         types `shouldBe` Just
           [aesonQQ|
             {
-              "format": "integer",
+              "format": "int32",
               "type": "integer"
             }
           |]
@@ -462,7 +462,7 @@ spec = describe "OpenAPI" $ do
         types `shouldBe` Just
           [aesonQQ|
             {
-              "format": "bigint",
+              "format": "int64",
               "type": "integer"
             }
           |]
@@ -850,7 +850,7 @@ spec = describe "OpenAPI" $ do
                 "type": "string"
               },
               {
-                "format": "integer",
+                "format": "int32",
                 "in": "query",
                 "name": "integer",
                 "required": false,
@@ -992,7 +992,7 @@ spec = describe "OpenAPI" $ do
                   "items": {}
                 },
                 "integer": {
-                  "format": "integer",
+                  "format": "int32",
                   "type": "integer"
                 },
                 "json": {

test/spec/Feature/Query/MultipleSchemaSpec.hs

@@ -281,7 +281,7 @@ spec =
                     "properties" : {
                       "id" : {
                         "description" : "Note:\nThis is a Primary Key.<pk/>",
-                        "format" : "integer",
+                        "format" : "int32",
                         "type" : "integer"
                       },
                       "name" : {
@@ -310,7 +310,7 @@ spec =
                     "properties" : {
                       "id" : {
                         "description" : "Note:\nThis is a Primary Key.<pk/>",
-                        "format" : "integer",
+                        "format" : "int32",
                         "type" : "integer"
                       },
                       "name" : {
@@ -339,7 +339,7 @@ spec =
                     "properties" : {
                       "id" : {
                         "description" : "Note:\nThis is a Primary Key.<pk/>",
-                        "format" : "integer",
+                        "format" : "int32",
                         "type" : "integer"
                       },
                       "name" : {
@@ -368,7 +368,7 @@ spec =
                     "properties" : {
                       "id" : {
                         "description" : "Note:\nThis is a Primary Key.<pk/>",
-                        "format" : "integer",
+                        "format" : "int32",
                         "type" : "integer"
                       },
                       "another_value" : {

test/spec/Main.hs

@@ -83,13 +83,12 @@ main = do
 
   -- cached schema cache so most tests run fast
   baseSchemaCache <- loadSCache pool testCfg
-  sockets <- AppState.initSockets testCfg
   loggerState <- Logger.init
   metricsState <- Metrics.init (configDbPoolSize testCfg)
 
   let
     initApp sCache st config = do
-      appState <- AppState.initWithPool sockets pool config loggerState metricsState (Metrics.observationMetrics metricsState)
+      appState <- AppState.initWithPool pool config loggerState metricsState (Metrics.observationMetrics metricsState)
       AppState.putPgVersion appState actualPgVersion
       AppState.putSchemaCache appState (Just sCache)
       return (st, postgrest (configLogLevel config) appState (pure ()))