federator-client: Use a new http2 connection on every request (#3602)

changelog.d/3-bug-fixes/WPB-4787

@@ -0,0 +1 @@
+Create a new http2 connection in every federator client request instead of using a shared connection.

libs/http2-manager/src/HTTP2/Client/Manager.hs

@@ -15,6 +15,10 @@ module HTTP2.Client.Manager
     ConnectionAlreadyClosed (..),
     disconnectTarget,
     disconnectTargetWithTimeout,
+    startPersistentHTTP2Connection,
+    sendRequestWithConnection,
+    HTTP2Conn (..),
+    ConnectionAction (..),
   )
 where
 

libs/wire-api-federation/default.nix

@@ -6,6 +6,7 @@
 , aeson
 , aeson-pretty
 , amqp
+, async
 , base
 , bytestring
 , bytestring-conversion
@@ -51,6 +52,7 @@ mkDerivation {
   libraryHaskellDepends = [
     aeson
     amqp
+    async
     base
     bytestring
     bytestring-conversion

libs/wire-api-federation/src/Wire/API/Federation/Client.hs

@@ -32,6 +32,7 @@ module Wire.API.Federation.Client
   )
 where
 
+import Control.Concurrent.Async
 import Control.Exception qualified as E
 import Control.Monad.Catch
 import Control.Monad.Codensity
@@ -58,6 +59,7 @@ import Network.HTTP.Media qualified as HTTP
 import Network.HTTP.Types qualified as HTTP
 import Network.HTTP2.Client qualified as HTTP2
 import Network.Wai.Utilities.Error qualified as Wai
+import OpenSSL.Session qualified as SSL
 import Servant.Client
 import Servant.Client.Core
 import Servant.Types.SourceT
@@ -113,13 +115,26 @@ liftCodensity = FederatorClient . lift . lift . lift
 headersFromTable :: HTTP2.HeaderTable -> [HTTP.Header]
 headersFromTable (headerList, _) = flip map headerList $ first HTTP2.tokenKey
 
+-- This opens a new http2 connection. Using a http2-manager leads to this problem https://wearezeta.atlassian.net/browse/WPB-4787
+-- FUTUREWORK: Replace with H2Manager.withHTTP2Request once the bugs are solved.
+withNewHttpRequest :: H2Manager.Target -> HTTP2.Request -> (HTTP2.Response -> IO a) -> IO a
+withNewHttpRequest target req k = do
+  ctx <- SSL.context
+  let cacheLimit = 20
+      sslRemoveTrailingDot = False
+      tcpConnectionTimeout = 30_000_000
+  sendReqMVar <- newEmptyMVar
+  thread <- liftIO . async $ H2Manager.startPersistentHTTP2Connection ctx target cacheLimit sslRemoveTrailingDot tcpConnectionTimeout sendReqMVar
+  let newConn = H2Manager.HTTP2Conn thread (putMVar sendReqMVar H2Manager.CloseConnection) sendReqMVar
+  H2Manager.sendRequestWithConnection newConn req k
+
 performHTTP2Request ::
   Http2Manager ->
   H2Manager.Target ->
   HTTP2.Request ->
   IO (Either FederatorClientHTTP2Error (ResponseF Builder))
-performHTTP2Request mgr target req = try $ do
-  H2Manager.withHTTP2Request mgr target req $ consumeStreamingResponseWith $ \resp -> do
+performHTTP2Request _mgr target req = try $ do
+  withNewHttpRequest target req $ consumeStreamingResponseWith $ \resp -> do
     b <-
       fmap (fromRight mempty)
         . runExceptT
@@ -210,7 +225,7 @@ withHTTP2StreamingRequest successfulStatus req handleResponse = do
     either throwError pure <=< liftCodensity $
       Codensity $ \k ->
         E.catches
-          (H2Manager.withHTTP2Request (ceHttp2Manager env) (False, hostname, port) req' (consumeStreamingResponseWith (k . Right)))
+          (withNewHttpRequest (False, hostname, port) req' (consumeStreamingResponseWith (k . Right)))
           [ E.Handler $ k . Left . FederatorClientHTTP2Error,
             E.Handler $ k . Left . FederatorClientHTTP2Error . FederatorClientConnectionError,
             E.Handler $ k . Left . FederatorClientHTTP2Error . FederatorClientHTTP2Exception,

libs/wire-api-federation/wire-api-federation.cabal

@@ -82,6 +82,7 @@ library
   build-depends:
       aeson                  >=2.0.1.0
     , amqp
+    , async
     , base                   >=4.6     && <5.0
     , bytestring
     , bytestring-conversion