add: use SO_REUSEPORT on platform supporting it

Author: mkleczek

CHANGELOG.md

@@ -13,12 +13,17 @@ All notable changes to this project will be documented in this file. From versio
 - Optimize requests with `Prefer: count=exact` that do not use ranges or `db-max-rows` by @laurenceisla in #3957
   + Removed unnecessary double count when building the `Content-Range`.
 - Add config `client_error_verbosity` to customize error verbosity by @taimoorzaeem in #4088, #3980, #3824
+- Use SO_REUSEPORT on platforms supporting it by @mkleczek in #4703 #4694
 
 ### Changed
 
 - Log error when `db-schemas` config contains schema `pg_catalog` or `information_schema` by @taimoorzaeem in #4359
   + Now fails at startup. Prior to this, it failed with `PGRST205` on requests related to these schemas.
 
+### Fixed
+
+- Shutdown should wait for in flight requests by @mkleczek in #4702
+
 ## [14.6] - 2026-03-06
 
 ### Fixed

nix/overlays/haskell-packages.nix

@@ -78,6 +78,53 @@ let
       hasql-pool = lib.dontCheck prev.hasql-pool_1_0_1;
       hasql-transaction = lib.dontCheck prev.hasql-transaction_1_1_0_1;
       postgresql-binary = lib.dontCheck (lib.doJailbreak prev.postgresql-binary_0_13_1_3);
+
+      http2 =
+        prev.callHackageDirect
+          {
+            pkg = "http2";
+            ver = "5.4.0";
+            sha256 = "sha256-PeEWVd61bQ8G7LvfLeXklzXqNJFaAjE2ecRMWJZESPE=";
+          }
+          { };
+
+      http-semantics =
+        prev.callHackageDirect
+          {
+            pkg = "http-semantics";
+            ver = "0.4.0";
+            sha256 = "sha256-rh0z51EKvsu5rQd5n2z3fSRjjEObouNZSBPO9NFYOF0=";
+          }
+          { };
+
+      network-run =
+        prev.callHackageDirect
+          {
+            pkg = "network-run";
+            ver = "0.5.0";
+            sha256 = "sha256-vbXh+CzxDsGApjqHxCYf/ijpZtUCApFbkcF5gyN0THU=";
+          }
+          { };
+
+      time-manager =
+        prev.callHackageDirect
+          {
+            pkg = "time-manager";
+            ver = "0.2.4";
+            sha256 = "sha256-sAt/331YLQ2IU3z90aKYSq1nxoazv87irsuJp7ZG3pw=";
+          }
+          { };
+
+      warp =
+        lib.dontCheck (prev.callCabal2nixWithOptions "warp"
+          (super.fetchFromGitHub {
+            owner = "mkleczek";
+            repo = "wai";
+            rev = "e9c9e784dee1b54461b94089175a969eb647d262";
+            #sha256 = "sha256-AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=";
+            sha256 = "sha256-q8vgvsNeKM/fHp3H6+W/tR4HicioqeaU9Eo3rb13bLo=";
+          }) "--subpath=warp"
+          { });
     };
 in
 {

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
 
@@ -43,8 +44,8 @@ admin appState req respond  = do
       respond $ Wai.responseLBS (if isMainAppReachable then HTTP.status200 else HTTP.status500) [] mempty
     ["ready"] ->
       let
-        status | not isMainAppReachable = HTTP.status500
-               | isPending              = HTTP.status503
+        status | isPending              = HTTP.status503
+               | not isMainAppReachable = HTTP.status500
                | isLoaded               = HTTP.status200
                | otherwise              = HTTP.status500
       in

src/PostgREST/App.hs

@@ -22,6 +22,7 @@ import GHC.IO.Exception (IOErrorType (..))
 import System.IO.Error  (ioeGetErrorType)
 
 import Control.Monad.Except     (liftEither)
+import Control.Monad.Extra      (whenJust)
 import Data.Either.Combinators  (mapLeft, whenLeft)
 import Data.Maybe               (fromJust)
 import Data.String              (IsString (..))
@@ -60,31 +61,50 @@ 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 (HostPreference,
+                                         bindPortGenEx)
+import qualified Data.Text              as T
+import qualified Network.HTTP.Types     as HTTP
+import qualified Network.Socket         as NS
+import           PostgREST.Unix         (createAndBindDomainSocketNoListen)
+import           Protolude              hiding (Handler)
 
 type Handler = ExceptT Error
 
 run :: AppState -> IO ()
 run appState = do
   conf@AppConfig{..} <- AppState.getConfig appState
 
-  AppState.schemaCacheLoader appState -- Loads the initial SchemaCache
-  Unix.installSignalHandlers (AppState.getMainThreadId appState) (AppState.schemaCacheLoader appState) (AppState.readInDbConfig False appState)
+  mainSocket <- initServerSocket conf
+  adminSocket <- initAdminServerSocket conf
+  let closeSockets = do
+        whenJust adminSocket NS.close
+        NS.close mainSocket
+  Unix.installSignalHandlers closeSockets (AppState.schemaCacheLoader appState) (AppState.readInDbConfig False appState)
+
+  Admin.runAdmin appState adminSocket mainSocket (serverSettings conf)
 
   Listener.runListener appState
 
-  Admin.runAdmin appState (serverSettings conf)
+  -- Kick off and wait for the initial SchemaCache load before listening on
+  -- the main API socket.
+  AppState.schemaCacheLoader appState
+  AppState.waitForSchemaCacheLoaded appState
+  void $ listenSocket mainSocket
 
   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
+  -- Hardcoding maximum graceful shutdown timeout (arbitrary set to 5 seconds)
+  -- This is unfortunate but necessary becase graceful shutdowns don't work with HTTP keep-alive
+  -- causing Warp to handle requests on already opened connections even if the listen socket is closed
+  -- See: https://github.com/yesodweb/wai/issues/853
+  Warp.runSettingsSocket (serverSettings conf & setOnException onWarpException) mainSocket app
   where
     observer = AppState.getObserver appState
 
@@ -229,3 +249,35 @@ addRetryHint delay response = do
 
 isServiceUnavailable :: Wai.Response -> Bool
 isServiceUnavailable response = Wai.responseStatus response == HTTP.status503
+
+initServerSocket :: AppConfig -> IO NS.Socket
+initServerSocket AppConfig{..} = case configServerUnixSocket 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 -> createAndBindDomainSocketNoListen path configServerUnixSocketMode
+  Nothing ->
+    bindPortTCPWithoutListen configServerPort (fromString $ T.unpack configServerHost)
+
+initAdminServerSocket :: AppConfig -> IO (Maybe NS.Socket)
+initAdminServerSocket AppConfig{..} =
+  traverse (`bindPortTCPWithReusePort` adminHost) configAdminServerPort
+  where
+    adminHost = fromString $ T.unpack configAdminServerHost
+
+bindPortTCPWithReusePort :: Int -> HostPreference -> IO NS.Socket
+bindPortTCPWithReusePort port hostPreference
+  = bindPortTCPWithoutListen port hostPreference >>= listenSocket
+
+bindPortTCPWithoutListen :: Int -> HostPreference -> IO NS.Socket
+bindPortTCPWithoutListen port hostPreference = do
+  -- Some unix variants can expose ReusePort but reject it at runtime.
+  -- Fall back to binding without ReusePort when that happens.
+  socketWithReusePort <- try (bindPortGenEx reusePortOpts NS.Stream port hostPreference) :: IO (Either SomeException NS.Socket)
+  either (const $ bindPortGenEx [] NS.Stream port hostPreference) pure socketWithReusePort
+  where
+    reusePortOpts = [(NS.ReusePort, 1)]
+
+listenSocket :: NS.Socket -> IO NS.Socket
+listenSocket sock = do
+  NS.listen sock (max 2048 NS.maxListenQueue)
+  pure sock