add: use SO_REUSEPORT on platform supporting it

CHANGELOG.md

@@ -15,6 +15,7 @@ All notable changes to this project will be documented in this file. From versio
 - Log schema cache queries timings on `log-level=debug` by @steve-chavez in #4805
 - Add GHC runtime metrics to the metrics endpoint by @mkleczek in #4862
 - Support running the admin server on a unix socket by @wolfgangwalther in #5003
+- Add config `server-reuseport` to allow starting multiple PostgREST instances using the same port on supported platforms by @mkleczek in #4703, #4694
 
 ### Fixed
 

docs/postgrest.dict

@@ -142,6 +142,7 @@ Redux
 refactor
 reloadable
 Reloadable
+reuseport
 requester's
 RESTful
 RLS

docs/references/admin_server.rst

@@ -5,6 +5,8 @@ Admin Server
 
 PostgREST provides an admin server that can be enabled by setting :ref:`admin-server-port` or `:ref:`admin-server-unix-socket`.
 
+Multiple PostgREST instances can share the same public API host and port when :ref:`server-reuseport` is enabled. Admin ports are not shared: give each instance a different :ref:`admin-server-port`, otherwise the new instance will fail to start.
+
 .. _health_check:
 
 Health Check

docs/references/configuration.rst

@@ -943,6 +943,50 @@ server-port
 
   The TCP port to bind the web server. Use ``0`` to automatically assign a port.
 
+.. _server-reuseport:
+
+server-reuseport
+----------------
+
+  =============== =================================
+  **Type**        Bool
+  **Default**     false
+  **Reloadable**  N
+  **Environment** PGRST_SERVER_REUSEPORT
+  **In-Database** `n/a`
+  =============== =================================
+
+  Enables ``SO_REUSEPORT`` on the TCP server socket. This allows multiple
+  PostgREST processes to bind to the same :ref:`server-host` and
+  :ref:`server-port` when the operating system supports it.
+  
+  For example, two PostgREST processes can use the same configuration:
+
+  .. code:: ini
+
+    server-host = "127.0.0.1"
+    server-port = 3000
+    server-reuseport = true
+
+  New connections are then distributed by the operating system between the
+  running PostgREST processes. This can be used to start a replacement process
+  before stopping the old one, or to run several PostgREST processes behind one
+  port.
+
+  If ``server-reuseport`` is disabled, starting another PostgREST process on
+  the same host and port will fail with the usual address-in-use error.
+
+  Enabling this setting on an operating system that does not support
+  ``SO_REUSEPORT`` is a configuration error. PostgREST will fail to start
+  instead of falling back to a normal TCP socket.
+
+  When running multiple PostgREST instances on the same :ref:`server-port`, use
+  a different ``admin-server-port`` for each instance. Admin ports are not shared
+  between instances, so readiness checks always target one specific PostgREST
+  instance.
+
+  This setting does not apply when :ref:`server-unix-socket` is used.
+
 .. _server-trace-header:
 
 server-trace-header

src/library/PostgREST/App.hs

@@ -68,7 +68,9 @@ import PostgREST.Version              (docsVersion, prettyVersion)
 import           Control.Monad.Writer
 import qualified Data.ByteString.Char8     as BS
 import qualified Data.List                 as L
-import           Data.Streaming.Network    (bindPortTCP)
+import           Data.Streaming.Network    (HostPreference,
+                                            bindPortGenEx,
+                                            bindPortTCP)
 import qualified Data.Text                 as T
 import qualified Network.HTTP.Types        as HTTP
 import           Network.HTTP.Types.Header (hVary, hWarning)
@@ -81,7 +83,7 @@ import System.Directory (doesPathExist)
 
 run :: AppState -> Weak ThreadId -> IO ()
 run appState mainThreadIdRef = do
-  conf <- AppState.getConfig appState
+  conf@AppConfig{configServerReusePort} <- AppState.getConfig appState
 
   mainSocketRef <- newIORef Nothing
   let setMainSocketRef = atomicWriteIORef mainSocketRef . Just
@@ -101,7 +103,10 @@ run appState mainThreadIdRef = do
     -- Kick off and wait for the initial SchemaCache load before creating the
     -- main API socket.
     AppState.schemaCacheLoader appState
-    AppState.waitForSchemaCacheInit appState
+    if configServerReusePort then
+      AppState.waitForSchemaCacheLoaded appState
+    else
+      AppState.waitForSchemaCacheInit appState
 
     bracket (initServerSocket conf) NS.close $ \mainSocket -> do
 
@@ -296,11 +301,11 @@ addRetryHint delay response = do
 isServiceUnavailable :: Wai.Response -> Bool
 isServiceUnavailable response = Wai.responseStatus response == HTTP.status503
 
-initSocket :: (Applicative f, Traversable f) => Maybe String -> FileMode -> Text -> f Int -> IO (f NS.Socket)
-initSocket unixSocket unixSocketMode tcpHost tcpPort =
+initSocket :: (Applicative f, Traversable f) => Maybe String -> FileMode -> Text -> f Int -> (Int -> HostPreference -> IO NS.Socket) -> IO (f NS.Socket)
+initSocket unixSocket unixSocketMode tcpHost tcpPort bindTCP =
   maybe initTCPSocket initDomainSocket unixSocket
   where
-    initTCPSocket = traverse (`bindPortTCP` (fromString $ T.unpack tcpHost)) tcpPort
+    initTCPSocket = traverse (`bindTCP` (fromString $ T.unpack tcpHost)) tcpPort
     -- 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
     initDomainSocket = fmap pure . (`createAndBindDomainSocket` unixSocketMode)
@@ -310,12 +315,20 @@ initServerSocket AppConfig{..} =
   runIdentity <$> initSocket
     configServerUnixSocket configServerUnixSocketMode
     configServerHost (pure configServerPort)
+    (if configServerReusePort then bindPortTCPWithReusePort else bindPortTCP)
 
 initAdminServerSocket :: AppConfig -> IO (Maybe NS.Socket)
 initAdminServerSocket AppConfig{..} =
   initSocket
     configAdminServerUnixSocket configAdminServerUnixSocketMode
     configAdminServerHost configAdminServerPort
+    bindPortTCP
+
+bindPortTCPWithReusePort :: Int -> HostPreference -> IO NS.Socket
+bindPortTCPWithReusePort port hostPreference =
+  bindPortGenEx [(NS.ReusePort, 1)] NS.Stream port hostPreference >>= listenSocket
+  where
+    listenSocket sock = NS.listen sock (max 2048 NS.maxListenQueue) $> sock
 
 checkMainAppLive :: IO (Maybe NS.Socket) -> Weak ThreadId -> IO Bool
 checkMainAppLive getMainSocket mainThreadIdRef =

src/library/PostgREST/AppState.hs

@@ -26,6 +26,7 @@ module PostgREST.AppState
   , isLoaded
   , isPending
   , waitForSchemaCacheInit
+  , waitForSchemaCacheLoaded
   ) where
 
 import qualified Data.ByteString.Char8      as BS
@@ -387,6 +388,9 @@ isSchemaCacheLoaded = atomically . (pure . fromMaybe False <=< tryReadTMVar) . g
 waitForSchemaCacheInit :: AppState -> IO ()
 waitForSchemaCacheInit = atomically . void . readTMVar . getSCStatusTMVar . stateSCacheStatus
 
+waitForSchemaCacheLoaded :: AppState -> IO ()
+waitForSchemaCacheLoaded = atomically . (check <=< readTMVar) . getSCStatusTMVar . stateSCacheStatus
+
 -- | Reads the in-db config and reads the config file again
 -- | We don't retry reading the in-db config after it fails immediately, because it could have user errors. We just report the error and continue.
 readInDbConfig :: Bool -> AppState -> IO ()

src/library/PostgREST/Config.hs

@@ -117,6 +117,7 @@ data AppConfig = AppConfig
   , configServerCorsAllowedOrigins  :: [Text]
   , configServerHost                :: Text
   , configServerPort                :: Int
+  , configServerReusePort           :: Bool
   , configServerTraceHeader         :: Maybe (CI.CI BS.ByteString)
   , configServerTimingEnabled       :: Bool
   , configServerUnixSocket          :: Maybe FilePath
@@ -204,6 +205,7 @@ toText conf =
       ,("server-cors-allowed-origins", q . T.intercalate "," . configServerCorsAllowedOrigins)
       ,("server-host",               q . configServerHost)
       ,("server-port",                   show . configServerPort)
+      ,("server-reuseport",              T.toLower . show . configServerReusePort)
       ,("server-trace-header",       q . T.decodeUtf8 . maybe mempty CI.original . configServerTraceHeader)
       ,("server-timing-enabled",         T.toLower . show . configServerTimingEnabled)
       ,("server-unix-socket",        q . maybe mempty T.pack . configServerUnixSocket)
@@ -323,6 +325,7 @@ parser optPath env dbSettings roleSettings roleIsolationLvl =
     <*> parseCORSAllowedOrigins "server-cors-allowed-origins"
     <*> (defaultServerHost <$> optString "server-host")
     <*> parseServerPort "server-port"
+    <*> (fromMaybe False <$> optBool "server-reuseport")
     <*> (fmap (CI.mk . encodeUtf8) <$> optString "server-trace-header")
     <*> (fromMaybe False <$> optBool "server-timing-enabled")
     <*> (fmap T.unpack <$> optString "server-unix-socket")
@@ -787,6 +790,7 @@ exampleConfigFile = S.unlines
   , ""
   , "server-host = \"!4\""
   , "server-port = 3000"
+  , "server-reuseport = false"
   , ""
   , "## Allow getting the request-response timing information through the `Server-Timing` header"
   , "server-timing-enabled = false"

test/io/configs/expected/aliases.config

@@ -38,6 +38,7 @@ openapi-server-proxy-uri = ""
 server-cors-allowed-origins = ""
 server-host = "!4"
 server-port = 3000
+server-reuseport = false
 server-timing-enabled = false
 server-trace-header = ""
 server-unix-socket = ""

test/io/configs/expected/boolean-numeric.config

@@ -38,6 +38,7 @@ openapi-server-proxy-uri = ""
 server-cors-allowed-origins = ""
 server-host = "!4"
 server-port = 3000
+server-reuseport = false
 server-timing-enabled = false
 server-trace-header = ""
 server-unix-socket = ""

test/io/configs/expected/boolean-string.config

@@ -38,6 +38,7 @@ openapi-server-proxy-uri = ""
 server-cors-allowed-origins = ""
 server-host = "!4"
 server-port = 3000
+server-reuseport = false
 server-timing-enabled = false
 server-trace-header = ""
 server-unix-socket = ""

test/io/configs/expected/defaults.config

@@ -38,6 +38,7 @@ openapi-server-proxy-uri = ""
 server-cors-allowed-origins = ""
 server-host = "!4"
 server-port = 3000
+server-reuseport = false
 server-timing-enabled = false
 server-trace-header = ""
 server-unix-socket = ""

test/io/configs/expected/no-defaults-with-db-other-authenticator.config

@@ -40,6 +40,7 @@ openapi-server-proxy-uri = "https://otherexample.org/api"
 server-cors-allowed-origins = "http://otherorigin.com"
 server-host = "0.0.0.0"
 server-port = 80
+server-reuseport = true
 server-timing-enabled = true
 server-trace-header = "traceparent"
 server-unix-socket = "/tmp/pgrst_io_test.sock"

test/io/configs/expected/no-defaults-with-db.config

@@ -40,6 +40,7 @@ openapi-server-proxy-uri = "https://example.org/api"
 server-cors-allowed-origins = "http://origin.com"
 server-host = "0.0.0.0"
 server-port = 80
+server-reuseport = true
 server-timing-enabled = false
 server-trace-header = "CF-Ray"
 server-unix-socket = "/tmp/pgrst_io_test.sock"

test/io/configs/expected/no-defaults.config

@@ -40,6 +40,7 @@ openapi-server-proxy-uri = "https://postgrest.org"
 server-cors-allowed-origins = "http://example.com"
 server-host = "0.0.0.0"
 server-port = 80
+server-reuseport = true
 server-timing-enabled = true
 server-trace-header = "X-Request-Id"
 server-unix-socket = "/tmp/pgrst_io_test.sock"

test/io/configs/expected/types.config

@@ -39,6 +39,7 @@ openapi-server-proxy-uri = ""
 server-cors-allowed-origins = ""
 server-host = "!4"
 server-port = 3000
+server-reuseport = false
 server-timing-enabled = false
 server-trace-header = ""
 server-unix-socket = ""

test/io/configs/expected/utf-8.config

@@ -38,6 +38,7 @@ openapi-server-proxy-uri = ""
 server-cors-allowed-origins = ""
 server-host = "!4"
 server-port = 3000
+server-reuseport = false
 server-timing-enabled = false
 server-trace-header = ""
 server-unix-socket = ""

test/io/configs/no-defaults-env.yaml

@@ -37,6 +37,7 @@ PGRST_OPENAPI_SERVER_PROXY_URI: 'https://postgrest.org'
 PGRST_SERVER_CORS_ALLOWED_ORIGINS: "http://example.com"
 PGRST_SERVER_HOST: 0.0.0.0
 PGRST_SERVER_PORT: 80
+PGRST_SERVER_REUSEPORT: true
 PGRST_SERVER_TRACE_HEADER: X-Request-Id
 PGRST_SERVER_TIMING_ENABLED: true
 PGRST_SERVER_UNIX_SOCKET: /tmp/pgrst_io_test.sock

test/io/configs/no-defaults.config

@@ -34,6 +34,7 @@ openapi-server-proxy-uri = "https://postgrest.org"
 server-cors-allowed-origins = "http://example.com"
 server-host = "0.0.0.0"
 server-port = 80
+server-reuseport = true
 server-trace-header = "X-Request-Id"
 server-timing-enabled = true
 server-unix-socket = "/tmp/pgrst_io_test.sock"

test/io/postgrest.py

@@ -266,7 +266,7 @@ def wait_until_status_code(url, max_seconds, status_code):
 
         time.sleep(0.1)
 
-    if response:
+    if response is not None:
         raise PostgrestTimedOut(f"{response.status_code}: {response.text}")
     else:
         raise PostgrestTimedOut()

test/io/test_io.py

@@ -19,6 +19,7 @@ from util import (
 )
 from postgrest import (
     Admin,
+    PostgrestTimedOut,
     freeport,
     is_ipv6,
     reset_statement_timeout,
@@ -176,7 +177,6 @@ def test_random_port_bound(defaultenv):
         assert True  # liveness check is done by run(), so we just need to check that it doesn't fail
 
 
-@pytest.mark.xfail(reason="PostgREST should not start on a used port", strict=True)
 def test_so_reuseport_zero_downtime_handover(defaultenv):
     "A second PostgREST instance should take over on the same main/admin ports without request failures."
 
@@ -204,7 +204,7 @@ def test_so_reuseport_zero_downtime_handover(defaultenv):
     # 6. Stop second PostgREST instance
     # 7. Verify client did not get any errors
     with run(
-        env={**defaultenv},
+        env={**defaultenv, "PGRST_SERVER_REUSEPORT": "true"},
         port=port,
         host=host,
         admin_port=admin_port,
@@ -226,10 +226,11 @@ def test_so_reuseport_zero_downtime_handover(defaultenv):
         try:
             time.sleep(1)
             with run(
-                env={**defaultenv},
+                env={**defaultenv, "PGRST_SERVER_REUSEPORT": "true"},
                 port=port,
                 host=host,
-                admin_port=admin_port,
+                # we do not set SO_REUSEPORT on admin socket
+                admin_port=freeport(used_ports=[port, admin_port]),
             ):
                 time.sleep(1)
                 first.process.terminate()
@@ -243,6 +244,30 @@ def test_so_reuseport_zero_downtime_handover(defaultenv):
     assert failures == []
 
 
+def test_so_reuseport_defaults_to_false(defaultenv):
+    "A second PostgREST instance should not bind to the same port by default."
+
+    host = "0.0.0.0"
+    port = freeport()
+    admin_port = freeport(used_ports=[port])
+
+    with run(
+        env={**defaultenv},
+        port=port,
+        host=host,
+        admin_port=admin_port,
+    ):
+        with pytest.raises(PostgrestTimedOut):
+            with run(
+                env={**defaultenv},
+                port=port,
+                host=host,
+                admin_port=freeport(used_ports=[port, admin_port]),
+                wait_max_seconds=1,
+            ):
+                pass
+
+
 def test_app_settings_reload(tmp_path, defaultenv):
     "App settings should be reloaded from file when PostgREST is sent SIGUSR2."
     config = (CONFIGSDIR / "sigusr2-settings.config").read_text()

test/observability/ObsHelper.hs

@@ -108,6 +108,7 @@ baseCfg = let secret = encodeUtf8 "reallyreallyreallyreallyverysafe" in
   , configServerCorsAllowedOrigins  = []
   , configServerHost                = "localhost"
   , configServerPort                = 3000
+  , configServerReusePort           = False
   , configServerTraceHeader         = Nothing
   , configServerUnixSocket          = Nothing
   , configServerUnixSocketMode      = 432

test/spec/SpecHelper.hs

@@ -172,6 +172,7 @@ baseCfg = let secret = encodeUtf8 "reallyreallyreallyreallyverysafe" in
   , configServerCorsAllowedOrigins  = []
   , configServerHost                = "localhost"
   , configServerPort                = 3000
+  , configServerReusePort           = False
   , configServerTraceHeader         = Nothing
   , configServerUnixSocket          = Nothing
   , configServerUnixSocketMode      = 432