diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,25 @@ # Revision history for grapesy +## 1.1.1 -- 2025-10-09 + +* Support `openConnection/closeConnection` + +## 1.1.0 -- 2025-07-17 + +* User-specified actions on connection/disconnect/reconnect [#280] + - Overhaul `ReconnectPolicy` to allow making `ReconnectDecision`s after + running some `IO` action. + - Introduce `ReconnectDecision`, which specifies the `ReconnectTo` target, the + `OnConnection` action to run upon reconnection, and the next + `ReconnectPolicy`. + - Add `OnConnection` actions to `ConnParams` and `ReconnectDecision` so that + users can track whether a connection is actually connected. +* `Network.GRPC.Common.Protobuf` does not import from `Data.ProtoLens.Labels`, + to avoid problems with `lens` [#283, Leonid Onokhov]. + NOTE: Users who want to rely on this will also need + https://github.com/google/proto-lens/pull/515. +* Lower bound on `http2-tls` changed to 0.4.9 [#289] + ## 1.0.1 -- 2025-04-01 * Support unix sockets [#275, Sjoerd Visscher] diff --git a/data/grpc-demo.key b/data/grpc-demo.key --- a/data/grpc-demo.key +++ b/data/grpc-demo.key @@ -1,28 +1,28 @@ -----BEGIN PRIVATE KEY----- -MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCnQR2GHEAcQIoK -LL0AdELGNoxw8UFIGLv3QHJH21znxXYL5RPNQc8MC0eoa/QTHHDTvuJ8C6l/7q0D -CDGdDF/gGBN0fSRsTQNPlC8RwdXI6ulw0jVPmjlTtAb1+L3k6AEV0smv87EPcLpv -zCpsZbH9Ouew1l+WCP0unTR5GYw4Zdv9vWMHdrIGPmpGwZn3hfALKkrHOsrrdb5n -iYa05JQi+YcfNmYF8gmSR8GA+2A3VL0itKN7/caBXuBKcvsBkExG/Ov1Vya+pdWc -+N+m1eazBjVm0VarZLv4DMcWNTWed1EULe52a8KAo14FlEDqhJsmmEJiZvzka5TL -z5stsn7rAgMBAAECggEAUgqDw+wJmpIh5BnL3/QnaPkK7L+6qPXRBdhr9klpCht2 -6yDEFNPqDttdnATQJau2wHcKu5Qw4Zse7LTROVr/kHne2S4ldqZUMG3cpNYy2qo4 -Neo20kQxSJivLWqFI0qWdbD+07syqANAwGQijydXJoMFcV3GZ18jagEc4yYf/O1W -+C8ooKg8/ThBaeUcRYpFWmBckgcvGFcVWMTjweoDZrzC0Bd+WhPykDlM/B6m/LjB -8kn7goV5iPSD8RICooJocLN2Nc2NW+LXrUw5VY3nlbLRGHDs4Qe30bJoU2b6PyXv -NDpEbOZrgERoopsI76dpCeDQ/bp06mHTEYEqrghYuQKBgQDQ4qWGkfkG3T5na+u1 -4t5ME+3mwtW34Te/wwUVUStzdg7IelzLx0+qzfINTTW1pvj93gbeVK1vGvTzzIE9 -2AdwxjffV5/woNUMH3vTr1qPRkJvJu+xwneu9PKyLhQHfXMkD0p6jltjBmbvBRCE -KB+5WYc9FEA8kP44RFOcOPXzvQKBgQDM+qIfn7bI+LdEq9jgOqr6XyUVCnT58W9I -ZJ30GkdDoP6TwxnAlX8wzwXURH+KIDxJTD6eS8lSOnaYK668sDIWo3OBLK1Fkemi -7BfZvkNDEtKcpYa9PQv5VtPUBbghK+ae7TGNge8xtMy9TmIu4rT05ynsYibjtWlB -trq8/EcTxwKBgBvWeLTMc2GkzpI94bXlvDZrWYMtaAoPa7yUovLKVH4Yt7OkCwXl -VAqxU5bOOWAyFnDOzB+JLWvnLcnn8TlqtuMip4OOS/Rnmrz43SnC7tC1Tlk92SfZ -gNXCMy3n0ieFYnjlyMk4e5lg2wrzo9XY+xFaixlqv3zS3e5lvLbPKIgJAoGBAIep -zgleHIzQyAMENaraSXUh6ZoObLNMDtn79eqsRcRF1pgXRYEHsMGuEu6VU1Ao252r -f7om8JyiowE90A2EE/KVxYmV9ywXUWmKFpL/cOcAmzIf/5hZwgYJaHNoQaB6vM0s -sWI1wAjG38bfDO55D0kTgdS4dYK5+2sJtHgGBEjbAoGAe0F+xVCikKwhPWJM/GlC -heTQDapyOV4/TZE2K/VP+ViO4d5CUr6K5qGVtHDDu/XQTaTbiN5CuuP5lvoPwfPF -L/RULhCv5kdyrlsTZwBofuruxxU5RGPKRREMq5aXqqmQWru1SS05ekdAf8cNoXRL -m7oXwgfow5GNg6OjduxfhUI= +MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCsQo73KVoCJ8fh ++CwdxiXewRBtnd/SeyKJxsskM90YRfO9BFxODW9mFkEvQx8tui9BE5Kbw3wV4Kxk +lc4bnj38E3Q6zqo2nJavNtI2U6/Edm5/ZcFfHrpAw97wRxek7rabSke8bvYtrz3y +wZEwt9hIWddPOQi43rKGPpOCyaDNmZpne+BpJPhW7/VqqQebGyDXRBZ8Wl1y1b4n +nem6MZ66zwLr01MWIWXKk+Fgog3zZFX8W4TdMBX4aaYgs3pqQvBnaxWyxQkPnWAR +P4IrfZVimjR4yP8DbLeMKzvq21sLEBUSvbB5JLt/gNTrWBnBOIMZ8vMBTMJRKMnb +wGWoCINFAgMBAAECggEAFOO3aaKynxtK4ozRcMTkN8iq4Ngp2eED1bhtTxUZBUYK +Ykwik3aOoU8mlYAqykVPULF6cHg61n5Z+ZKvHWtJsgV77Vu9iYTgwxu/T0ZDxOvl +x35D/nB//rWiFfpRFDe8nkVaQLAmG3EqboNpw4Iv8MowUZOliqG6/YueINiprvTz +v3/T0u8Ry/oEFae/YNFME41Q+glYxOptEVkXMN89KccEl+W2eFNv3BdCukQKhpoE +fGR1+qRLv3uuRmDENDFw+Agp7XQ/9MucZSoklh1vuw1998p3JmaSlyurRoUDFiBE +YXRH/TFXoi6nitFdaGjh6K+1BekdZt+7bNoZZyvG+wKBgQDwMbWIvLeOgwbex4NN +gQk03WcvUpP56LrXrpkQ+JaO46FatengmGcTLPqQ56atNV0uNvHd8JwEYg2lBJWw +LSWhGV1Sa2CkXfp4LFgJ0Y3xxST7sfP4/HxnK3fg9f0nXOFmRdup96CI5LUVCtAC +jBbjwXQXy5rXy/4jY0k1YCjdAwKBgQC3mG+SGLm7GHp9GKCxXtXudg6vTBoUjHdh +NJSrVuRC2MRYmlZPVMYniDDHM/G8v+DzKCBPlGQPqWorRcvsD95xrOE7W+/zuiNT +lYSa9EKWolzq6aOAHNG2pBxVfZiW85IFw5GQLaCU3V4YP/oYeJa3eESxHyy+gPFN +6zOmMqw4FwKBgHzynuieoy3zYyOIzfkHYu6pLgAkCO477tY78UwuxMNYDpvNffhj +z1reTwoKN15rICnmUzOM8twk1cw98lBPa/+93hn92awnZyAUkUeqRxi54V89Vxjy +3xQcPKQ90o8jde1p8bcdJdmQf9KOaV6p2U5pWCb4t3gCmhV2lKK7fwZZAoGAB8Bg +3Ys7tEGJUmTKzBJT9/h2EEKnSzzPAYSlzkIh6wyZ5Z/GixzqLNscLBzuVOjJB5sn +GhUK0Hp3qBIPVQ0qeCQzcj0keWbffPTwH1a2xQNf5u8sXwlYdVyicZ2W5rCr9qBW +Mf8rK33ZLi7tUUEuI9rpE41cZ0KsbXzDtn2nNcsCgYBfqjgh8KOuE4firXNeMKBR +elhVFiCogJ8pzduq4yJBJs+PGwHuCH7JZNQTOVyrULJ43MhG0/xSf1EmraRKLOOK +DnqmdyNyeH0aZVMzsrzL+L5I7TPiu/3h6uKfrnloOJ7OBcg8lVEREhtDPfHY1s0M +bxpcu7Q5sa3qrG9pZ62DIw== -----END PRIVATE KEY----- diff --git a/data/grpc-demo.pem b/data/grpc-demo.pem --- a/data/grpc-demo.pem +++ b/data/grpc-demo.pem @@ -1,19 +1,19 @@ -----BEGIN CERTIFICATE----- -MIIDCTCCAfGgAwIBAgIUbt4YUTJ3u4cwD1ZrXl7AU792K5AwDQYJKoZIhvcNAQEL -BQAwFDESMBAGA1UEAwwJMTI3LjAuMC4xMB4XDTI0MDYxNDEwNDcxNVoXDTI1MDYx -NDEwNDcxNVowFDESMBAGA1UEAwwJMTI3LjAuMC4xMIIBIjANBgkqhkiG9w0BAQEF -AAOCAQ8AMIIBCgKCAQEAp0EdhhxAHECKCiy9AHRCxjaMcPFBSBi790ByR9tc58V2 -C+UTzUHPDAtHqGv0Exxw077ifAupf+6tAwgxnQxf4BgTdH0kbE0DT5QvEcHVyOrp -cNI1T5o5U7QG9fi95OgBFdLJr/OxD3C6b8wqbGWx/TrnsNZflgj9Lp00eRmMOGXb -/b1jB3ayBj5qRsGZ94XwCypKxzrK63W+Z4mGtOSUIvmHHzZmBfIJkkfBgPtgN1S9 -IrSje/3GgV7gSnL7AZBMRvzr9VcmvqXVnPjfptXmswY1ZtFWq2S7+AzHFjU1nndR -FC3udmvCgKNeBZRA6oSbJphCYmb85GuUy8+bLbJ+6wIDAQABo1MwUTAdBgNVHQ4E -FgQUM565CtSaHBunw9mQnq1/apbCsL4wHwYDVR0jBBgwFoAUM565CtSaHBunw9mQ -nq1/apbCsL4wDwYDVR0TAQH/BAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAQEAEiX4 -8xBfkSl6R5sTYpP0eRRTiG0x+tG1hsTLYWbsCQIiUi5gIZm6CnQ25NO4FJ4J623p -NIM1pfJP44J/7kOzhquLLBX5t9UjWsJdyh/IqRkj4Y3ZZqykLTUDOp9hpGRjXFnu -Fjas/CCfhfZ/wfvUPUhR1sq843Py7AMnz39OsC9etkGTTlL0QZCteQzDaZ+Yrwwx -SDMp1TpZQs6+7Gj8cvyrGt/GnCb8Wc6K03cYypdpVtTczl9ouxTXyyPeoVhKRCPv -Z0JyJMdFuRmoNPnUq8WR08IvKADURaspLbv39X1rAOj8zhEWGwngLQ9LqF1jz4ee -BOkxakzQa4lbLofyYQ== +MIIDCTCCAfGgAwIBAgIUJUaCdYsL0XZo/YVU22eiVgsCt5UwDQYJKoZIhvcNAQEL +BQAwFDESMBAGA1UEAwwJMTI3LjAuMC4xMB4XDTI1MDcxNTEyMzIzM1oXDTI2MDcx +NTEyMzIzM1owFDESMBAGA1UEAwwJMTI3LjAuMC4xMIIBIjANBgkqhkiG9w0BAQEF +AAOCAQ8AMIIBCgKCAQEArEKO9ylaAifH4fgsHcYl3sEQbZ3f0nsiicbLJDPdGEXz +vQRcTg1vZhZBL0MfLbovQROSm8N8FeCsZJXOG549/BN0Os6qNpyWrzbSNlOvxHZu +f2XBXx66QMPe8EcXpO62m0pHvG72La898sGRMLfYSFnXTzkIuN6yhj6TgsmgzZma +Z3vgaST4Vu/1aqkHmxsg10QWfFpdctW+J53pujGeus8C69NTFiFlypPhYKIN82RV +/FuE3TAV+GmmILN6akLwZ2sVssUJD51gET+CK32VYpo0eMj/A2y3jCs76ttbCxAV +Er2weSS7f4DU61gZwTiDGfLzAUzCUSjJ28BlqAiDRQIDAQABo1MwUTAdBgNVHQ4E +FgQUjOfzmF9JJNoZcL82V79PWWy7qH8wHwYDVR0jBBgwFoAUjOfzmF9JJNoZcL82 +V79PWWy7qH8wDwYDVR0TAQH/BAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAQEAHqdY +wn+go05bD0I85owKiLvvFB5TROJXJCAJpEaXrDD16fbU28xLEvG44FqJ5/WRGUTy +odr3HOTlPuWkjlLk2y90m0NkVM6Jc+VS/HuAF26GYWGpB+cMsjjuXGInV4+Vy/BQ +1Yu/oH9tFMEtGDQHVaV/dYd2iORf/ANf2Iu9VAf2pZ2muFoWqd8LXvtrjWnkrQOk +jiGs7khc/H7apn7yx+7F3D2lHwK1eiZXJnqaTMyJfkcDFyphAMgp93kAt8lEJvU9 +DFIyFuybUtCCN4f5NAYi+p4GaOilgFYGpRYm9WntH5PiHSVTgibfVX/4uISHU16W +DXpIaptTShhjG8iXGQ== -----END CERTIFICATE----- diff --git a/grapesy.cabal b/grapesy.cabal --- a/grapesy.cabal +++ b/grapesy.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: grapesy -version: 1.0.1 +version: 1.1.1 synopsis: Native Haskell implementation of the gRPC framework description: This is a fully compliant and feature complete native Haskell implementation of gRPC, Google's RPC framework. @@ -156,7 +156,7 @@ , exceptions >= 0.10 && < 0.11 , grpc-spec >= 1.0 && < 1.1 , http-types >= 0.12 && < 0.13 - , http2-tls >= 0.4.5 && < 0.5 + , http2-tls >= 0.4.9 && < 0.5 , lens >= 5.0 && < 5.4 , mtl >= 2.2 && < 2.4 , network >= 3.2.4 && < 3.3 @@ -232,6 +232,7 @@ Test.Sanity.Disconnect Test.Sanity.EndOfStream Test.Sanity.Interop + Test.Sanity.NoIsLabel Test.Sanity.Reclamation Test.Sanity.StreamingType.CustomFormat Test.Sanity.StreamingType.NonStreaming @@ -266,6 +267,7 @@ , http2 , mtl , network + , proto-lens , proto-lens-protobuf-types , stm , text @@ -394,6 +396,7 @@ , grpc-spec , mtl , network + , proto-lens , text build-depends: @@ -433,6 +436,7 @@ , bytestring , containers , deepseq + , proto-lens , text , unordered-containers diff --git a/interop/Interop/Client/Connect.hs b/interop/Interop/Client/Connect.hs --- a/interop/Interop/Client/Connect.hs +++ b/interop/Interop/Client/Connect.hs @@ -1,5 +1,7 @@ module Interop.Client.Connect (testServer) where +import Data.ProtoLens.Labels () + import Network.GRPC.Client import Interop.Cmdline diff --git a/interop/Interop/Server/Common.hs b/interop/Interop/Server/Common.hs --- a/interop/Interop/Server/Common.hs +++ b/interop/Interop/Server/Common.hs @@ -8,6 +8,7 @@ ) where import Control.Exception +import Data.ProtoLens.Labels () import Network.GRPC.Common import Network.GRPC.Common.Protobuf diff --git a/interop/Interop/Server/PingService/Ping.hs b/interop/Interop/Server/PingService/Ping.hs --- a/interop/Interop/Server/PingService/Ping.hs +++ b/interop/Interop/Server/PingService/Ping.hs @@ -1,5 +1,7 @@ module Interop.Server.PingService.Ping (handle) where +import Data.ProtoLens.Labels () + import Network.GRPC.Common.Protobuf import Proto.Ping diff --git a/interop/Interop/Util/Messages.hs b/interop/Interop/Util/Messages.hs --- a/interop/Interop/Util/Messages.hs +++ b/interop/Interop/Util/Messages.hs @@ -11,6 +11,7 @@ import Data.ByteString qualified as BS.Strict import Data.ByteString qualified as Strict (ByteString) import Data.ByteString.Char8 qualified as BS.Strict.Char8 +import Data.ProtoLens.Labels () import Network.GRPC.Common.Protobuf diff --git a/kvstore/KVStore/API/Protobuf.hs b/kvstore/KVStore/API/Protobuf.hs --- a/kvstore/KVStore/API/Protobuf.hs +++ b/kvstore/KVStore/API/Protobuf.hs @@ -2,6 +2,8 @@ module KVStore.API.Protobuf (client, server) where +import Data.ProtoLens.Labels () + import Network.GRPC.Client (rpc) import Network.GRPC.Client qualified as Client import Network.GRPC.Client.StreamType.IO qualified as Client diff --git a/src/Network/GRPC/Client.hs b/src/Network/GRPC/Client.hs --- a/src/Network/GRPC/Client.hs +++ b/src/Network/GRPC/Client.hs @@ -1,14 +1,17 @@ -{-# LANGUAGE OverloadedStrings #-} - module Network.GRPC.Client ( -- * Connecting to the server Connection -- opaque , Server(..) , ConnParams(..) , withConnection + , openConnection + , closeConnection -- ** Reconnection policy , ReconnectPolicy(..) + , ReconnectDecision(..) + , Reconnect(..) + , OnConnection(..) , ReconnectTo(..) , exponentialBackoff diff --git a/src/Network/GRPC/Client/Connection.hs b/src/Network/GRPC/Client/Connection.hs --- a/src/Network/GRPC/Client/Connection.hs +++ b/src/Network/GRPC/Client/Connection.hs @@ -1,3 +1,4 @@ +{-# LANGUAGE CPP #-} {-# LANGUAGE OverloadedStrings #-} -- | Connection to a server @@ -10,12 +11,17 @@ -- * Definition Connection -- opaque , withConnection + , openConnection + , closeConnection -- * Configuration , Server(..) , ServerValidation(..) , SslKeyLog(..) , ConnParams(..) , ReconnectPolicy(..) + , ReconnectDecision(..) + , Reconnect(..) + , OnConnection(..) , ReconnectTo(..) , exponentialBackoff -- * Using the connection @@ -80,6 +86,9 @@ -- | Connection state , connStateVar :: TVar ConnectionState + + -- | Flag for garbage collection. + , connOutOfScope :: MVar () } {------------------------------------------------------------------------------- @@ -98,6 +107,9 @@ -- Individual RPC calls can override this through 'CallParams'. , connDefaultTimeout :: Maybe Timeout + -- | Action to run upon successful connection + , connOnConnection :: OnConnection + -- | Reconnection policy -- -- NOTE: The default 'ReconnectPolicy' is 'DontReconnect', as per the @@ -147,6 +159,7 @@ def = ConnParams { connCompression = def , connDefaultTimeout = Nothing + , connOnConnection = def , connReconnectPolicy = def , connContentType = Just ContentTypeDefault , connVerifyHeaders = False @@ -161,26 +174,59 @@ -- | Reconnect policy -- -- See 'exponentialBackoff' for a convenient function to construct a policy. -data ReconnectPolicy = +-- +-- When we get disconnected from the server, we will 'runReconnectPolicy' +-- to decide what to do next. This can run arbitrary IO actions; two example +-- use cases are +-- +-- * wait until we reconnect (run 'threadDelay') +-- * update some application-specific state to indicate that we are not +-- currently connected to the server (see also 'onReconnect') +newtype ReconnectPolicy = ReconnectPolicy{ + runReconnectPolicy :: IO ReconnectDecision + } + +-- | Decision on whether to reconnect or not +-- +-- See 'ReconnectPolicy'. +data ReconnectDecision = -- | Do not attempt to reconnect -- -- When we get disconnected from the server (or fail to establish a -- connection), do not attempt to connect again. DontReconnect - -- | Reconnect to the (potentially different) server after the IO action - -- returns - -- - -- The 'ReconnectTo' can be used to implement a rudimentary redundancy - -- scheme. For example, you could decide to reconnect to a known fallback - -- server after connection to a main server fails a certain number of times. - -- - -- This is a very general API: typically the IO action will call - -- 'threadDelay' after some amount of time (which will typically involve - -- some randomness), but it can be used to do things such as display a - -- message to the user somewhere that the client is reconnecting. - | ReconnectAfter ReconnectTo (IO ReconnectPolicy) + -- | Reconnect (optionally to a different server) + | DoReconnect Reconnect +-- | Decision made by a 'ReconnectPolicy' +data Reconnect = Reconnect { + -- | Server to reconnect to + -- + -- This can be used to implement a rudimentary redundancy scheme. For + -- example, you could decide to reconnect to a known fallback server after + -- connection to a main server fails a certain number of times. + reconnectTo :: ReconnectTo + + -- | Action to run on successful reconnect + -- + -- If it is 'Nothing', the 'connOnReconnect' given in the initial + -- 'ConnParams' will be used instead. + , onReconnect :: Maybe OnConnection + + -- | New policy + -- + -- If the /next/ connection attempt fails at any point, use this as the + -- next 'ReconnectPolicy' + , nextPolicy :: ReconnectPolicy + } + +-- | An action to run upon successful (re)connection to a server +-- +-- This can be used to, for example, display a message to the user that the +-- connection has been (re)established . +newtype OnConnection = OnConnection { runOnConnection :: IO () } + -- | What server should we attempt to reconnect to? -- -- * 'ReconnectToPrevious' will attempt to reconnect to the last server we @@ -198,10 +244,11 @@ -- The default follows the gRPC specification of Wait for Ready semantics -- . instance Default ReconnectPolicy where - def = DontReconnect + def = ReconnectPolicy $ pure DontReconnect -instance Default ReconnectTo where - def = ReconnectToPrevious +-- | The default 'OnConnection' is to do nothing +instance Default OnConnection where + def = OnConnection $ pure () -- | Exponential backoff -- @@ -229,11 +276,15 @@ exponentialBackoff waitFor e = go where go :: (Double, Double) -> Word -> ReconnectPolicy - go _ 0 = DontReconnect - go (lo, hi) n = ReconnectAfter def $ do + go _ 0 = ReconnectPolicy $ pure DontReconnect + go (lo, hi) n = ReconnectPolicy $ do delay <- randomRIO (lo, hi) waitFor $ round $ delay * 1_000_000 - return $ go (lo * e, hi * e) (pred n) + return $ DoReconnect Reconnect { + reconnectTo = ReconnectToOriginal + , onReconnect = Nothing + , nextPolicy = go (lo * e, hi * e) (pred n) + } {------------------------------------------------------------------------------- Fatal exceptions (no point reconnecting) @@ -266,7 +317,7 @@ Open a new connection -------------------------------------------------------------------------------} --- | Open connection to the server +-- | Open a connection to the server. -- -- See 'Network.GRPC.Client.withRPC' for making individual RPCs on the new -- connection. @@ -281,11 +332,11 @@ -- -- If the connection to the server is lost /after/ it has been established, any -- currently ongoing RPC calls will be closed; attempts at further communication --- on any of these calls will result in an exception being thrown. However, if --- the 'ReconnectPolicy' allows, we will automatically try to re-establish a --- connection to the server. This can be especially important when there is a --- proxy between the client and the server, which may drop an existing --- connection after a certain period. +-- on any of these calls will result in a 'ServerDisconnected' exception being +-- thrown. If that exception is caught, and the 'ReconnectPolicy' allows, we +-- will automatically try to re-establish a connection to the server. This can +-- be especially important when there is a proxy between the client and the +-- server, which may drop an existing connection after a certain period. -- -- NOTE: The /default/ 'ReconnectPolicy' is 'DontReconnect', as per the gRPC -- specification of "Wait for ready" semantics. You may wish to override this @@ -302,6 +353,22 @@ -> (Connection -> IO a) -> IO a withConnection connParams server k = do + bracket (openConnection connParams server) closeConnection k + +-- | Open a connection to the server. +-- +-- See 'withConnection' for details. +-- +-- __Warning:__ +-- Connections hold open resources and must be closed using 'closeConnection'. +-- To prevent resource and memory leaks due to asynchronous exceptions, it is +-- recommended to use the bracketed function 'withConnection' whenever +-- possible, and otherwise run functions that allocate and release a resource +-- with asynchronous exceptions masked, and ensure that every use allocate +-- operation is followed by the corresponding release operation even in the +-- presence of asynchronous exceptions, e.g., using 'bracket'. +openConnection :: ConnParams -> Server -> IO Connection +openConnection connParams server = do connMetaVar <- newMVar $ Meta.init (connInitCompression connParams) connStateVar <- newTVarIO ConnectionNotReady @@ -314,9 +381,12 @@ -- when we no longer need the connection (which we indicate by writing to -- connOutOfScope). void $ forkLabelled "grapesy:stayConnected" $ stayConnectedThread - k Connection {connParams, connMetaVar, connStateVar} - `finally` putMVar connOutOfScope () + pure Connection {connParams, connMetaVar, connStateVar, connOutOfScope} +-- | Close a connection to the server. +closeConnection :: Connection -> IO () +closeConnection conn = putMVar (connOutOfScope conn) () + {------------------------------------------------------------------------------- Making use of the connection -------------------------------------------------------------------------------} @@ -380,21 +450,27 @@ -- -- This is an internal data structure used only in 'stayConnected' and helpers. data Attempt = ConnectionAttempt { - attemptParams :: ConnParams - , attemptState :: TVar ConnectionState - , attemptOutOfScope :: MVar () - , attemptClosed :: TMVar (Maybe SomeException) + attemptParams :: ConnParams + , attemptOnConnection :: OnConnection + , attemptState :: TVar ConnectionState + , attemptOutOfScope :: MVar () + , attemptClosed :: TMVar (Maybe SomeException) } newConnectionAttempt :: ConnParams + -> OnConnection -> TVar ConnectionState -> MVar () -> IO Attempt -newConnectionAttempt attemptParams attemptState attemptOutOfScope = do +newConnectionAttempt attemptParams + attemptOnConnection + attemptState + attemptOutOfScope = do attemptClosed <- newEmptyTMVarIO return ConnectionAttempt{ attemptParams + , attemptOnConnection , attemptState , attemptOutOfScope , attemptClosed @@ -408,12 +484,15 @@ -> MVar () -> IO () stayConnected connParams initialServer connStateVar connOutOfScope = do - loop initialServer (connReconnectPolicy connParams) + loop + initialServer + (connOnConnection connParams) + (connReconnectPolicy connParams) where - loop :: Server -> ReconnectPolicy -> IO () - loop server remainingReconnectPolicy = do + loop :: Server -> OnConnection -> ReconnectPolicy -> IO () + loop server onConnection remainingReconnectPolicy = do -- Start new attempt (this just allocates some internal state) - attempt <- newConnectionAttempt connParams connStateVar connOutOfScope + attempt <- newConnectionAttempt connParams onConnection connStateVar connOutOfScope -- Just like in 'runHandler' on the server side, it is important that -- 'stayConnected' runs in a separate thread. If it does not, then the @@ -450,23 +529,32 @@ case mRes of Right () -> do atomically $ writeTVar connStateVar $ ConnectionOutOfScope - Left err -> do - case (isFatalException err, thisReconnectPolicy) of - (True, _) -> do - atomically $ writeTVar connStateVar $ ConnectionAbandoned err - (False, DontReconnect) -> do - atomically $ writeTVar connStateVar $ ConnectionAbandoned err + Left err + | isFatalException err -> atomically $ writeTVar connStateVar $ ConnectionAbandoned err - (False, ReconnectAfter to f) -> do - let - nextServer = - case to of - ReconnectToPrevious -> server - ReconnectToOriginal -> initialServer - ReconnectToNew new -> new + | otherwise -> do + -- Mark the connection as not ready /before/ running the reconnt + -- policy. This prevents any attempts to use the connection + -- while the policy is running. atomically $ writeTVar connStateVar $ ConnectionNotReady - loop nextServer =<< f + runReconnectPolicy thisReconnectPolicy >>= \case + DontReconnect -> do + atomically $ writeTVar connStateVar $ ConnectionAbandoned err + DoReconnect reconnect -> do + let + nextServer = + case reconnectTo reconnect of + ReconnectToPrevious -> server + ReconnectToOriginal -> initialServer + ReconnectToNew new -> new + onReconnect' = + case onReconnect reconnect of + Just act -> act + Nothing -> connOnConnection connParams + + loop nextServer onReconnect' $ nextPolicy reconnect + -- | Unix domain socket connection connectUnix :: ConnParams -> Attempt -> FilePath -> IO () connectUnix connParams attempt path = do @@ -500,6 +588,7 @@ atomically $ writeTVar (attemptState attempt) $ ConnectionReady (attemptClosed attempt) conn + runOnConnection $ attemptOnConnection attempt takeMVar $ attemptOutOfScope attempt where ConnParams{connHTTP2Settings} = connParams @@ -538,15 +627,14 @@ let settings :: HTTP2.TLS.Client.Settings settings = HTTP2.TLS.Client.defaultSettings { - HTTP2.TLS.Client.settingsValidateCert = + HTTP2.TLS.Client.settingsKeyLogger = keyLogger + , HTTP2.TLS.Client.settingsCAStore = caStore + , HTTP2.TLS.Client.settingsAddrInfoFlags = [] + + , HTTP2.TLS.Client.settingsValidateCert = case validation of ValidateServer _ -> True NoServerValidation -> False - - , HTTP2.TLS.Client.settingsCAStore = caStore - , HTTP2.TLS.Client.settingsKeyLogger = keyLogger - , HTTP2.TLS.Client.settingsAddrInfoFlags = [] - , HTTP2.TLS.Client.settingsOpenClientSocket = openClientSocket connHTTP2Settings , HTTP2.TLS.Client.settingsConcurrentStreams = fromIntegral $ @@ -573,6 +661,7 @@ atomically $ writeTVar (attemptState attempt) $ ConnectionReady (attemptClosed attempt) conn + runOnConnection $ attemptOnConnection attempt takeMVar $ attemptOutOfScope attempt where ConnParams{connHTTP2Settings} = connParams diff --git a/src/Network/GRPC/Common/Protobuf.hs b/src/Network/GRPC/Common/Protobuf.hs --- a/src/Network/GRPC/Common/Protobuf.hs +++ b/src/Network/GRPC/Common/Protobuf.hs @@ -35,8 +35,7 @@ import Data.Function ((&)) import Data.Int import Data.Maybe (fromMaybe) -import Data.ProtoLens.Field (HasField(..)) -import Data.ProtoLens.Labels () -- provides instances for OverloadedLabels +import Data.ProtoLens.Field (HasField(..), field) import Data.ProtoLens.Message (FieldDefault(..), Message(defMessage)) import Data.Text (Text) @@ -75,9 +74,9 @@ status :: Proto Status status = defMessage - & #code .~ fromIntegral (fromGrpcError protobufErrorCode) - & #message .~ fromMaybe fieldDefault protobufErrorMessage - & #details .~ protobufErrorDetails + & field @"code" .~ fromIntegral (fromGrpcError protobufErrorCode) + & field @"message" .~ fromMaybe fieldDefault protobufErrorMessage + & field @"details" .~ protobufErrorDetails -- | Variation of 'throwProtobufError' for a homogenous list of details -- @@ -101,11 +100,11 @@ } Just statusEnc -> do status :: Proto Status <- parseStatus statusEnc - protobufErrorCode <- checkErrorCode (status ^. #code) + protobufErrorCode <- checkErrorCode (status ^. field @"code") return ProtobufError{ protobufErrorCode - , protobufErrorMessage = constructErrorMessage (status ^. #message) - , protobufErrorDetails = status ^. #details + , protobufErrorMessage = constructErrorMessage (status ^. field @"message") + , protobufErrorDetails = status ^. field @"details" } where -- The gRPC specification mandates diff --git a/src/Network/GRPC/Util/Thread.hs b/src/Network/GRPC/Util/Thread.hs --- a/src/Network/GRPC/Util/Thread.hs +++ b/src/Network/GRPC/Util/Thread.hs @@ -152,7 +152,7 @@ threadBody label state body = do labelThisThread label threadId <- myThreadId - initState <- atomically $ readTVar state + initState <- readTVarIO state -- See discussion of 'ThreadNotStarted' -- It's critical that async exceptions are masked at this point. @@ -376,4 +376,3 @@ withoutDeadlockDetection k = do threadId <- myThreadId bracket (newStablePtr threadId) freeStablePtr $ \_ -> k - diff --git a/test-grapesy/Main.hs b/test-grapesy/Main.hs --- a/test-grapesy/Main.hs +++ b/test-grapesy/Main.hs @@ -22,6 +22,7 @@ import Test.Sanity.Disconnect qualified as Disconnect import Test.Sanity.EndOfStream qualified as EndOfStream import Test.Sanity.Interop qualified as Interop +import Test.Sanity.NoIsLabel qualified as NoIsLabel import Test.Sanity.Reclamation qualified as Reclamation import Test.Sanity.StreamingType.CustomFormat qualified as StreamingType.CustomFormat import Test.Sanity.StreamingType.NonStreaming qualified as StreamingType.NonStreaming @@ -43,6 +44,7 @@ , Interop.tests , Reclamation.tests , BrokenDeployments.tests + , NoIsLabel.tests ] , testGroup "Regression" [ Issue102.tests diff --git a/test-grapesy/Test/Driver/ClientServer.hs b/test-grapesy/Test/Driver/ClientServer.hs --- a/test-grapesy/Test/Driver/ClientServer.hs +++ b/test-grapesy/Test/Driver/ClientServer.hs @@ -35,6 +35,7 @@ import Control.Monad import Control.Monad.Catch import Control.Monad.IO.Class +import Data.ProtoLens.Labels () import Data.Text qualified as Text import Network.HTTP2.Server qualified as HTTP2.Server import Network.Socket (PortNumber) @@ -470,9 +471,15 @@ -- This avoids a race condition between the server starting first -- and the client starting first. , connReconnectPolicy = - Client.ReconnectAfter def $ do - threadDelay 100_000 - return Client.DontReconnect + Client.ReconnectPolicy $ do + threadDelay 100_000 + return $ + Client.DoReconnect Client.Reconnect { + reconnectTo = Client.ReconnectToOriginal + , onReconnect = Nothing + , nextPolicy = def + } + , connOnConnection = def } clientServer :: Client.Server diff --git a/test-grapesy/Test/Driver/Dialogue.hs b/test-grapesy/Test/Driver/Dialogue.hs --- a/test-grapesy/Test/Driver/Dialogue.hs +++ b/test-grapesy/Test/Driver/Dialogue.hs @@ -1,5 +1,3 @@ -{-# LANGUAGE OverloadedStrings #-} - module Test.Driver.Dialogue ( module X ) where diff --git a/test-grapesy/Test/Sanity/Disconnect.hs b/test-grapesy/Test/Sanity/Disconnect.hs --- a/test-grapesy/Test/Sanity/Disconnect.hs +++ b/test-grapesy/Test/Sanity/Disconnect.hs @@ -205,18 +205,32 @@ go :: Int -> Client.ReconnectPolicy go n | n == 5 - = Client.ReconnectAfter def $ do + = Client.ReconnectPolicy $ do killRestarted <- startServer port2 <- ipcRead putMVar signalRestart killRestarted - return $ - Client.ReconnectAfter - (Client.ReconnectToNew $ serverAddress port2) - (pure Client.DontReconnect) + return $ Client.DoReconnect Client.Reconnect { + Client.nextPolicy = + Client.ReconnectPolicy $ + pure $ Client.DoReconnect Client.Reconnect { + Client.reconnectTo = + Client.ReconnectToNew $ serverAddress port2 + , Client.nextPolicy = + Client.ReconnectPolicy $ pure Client.DontReconnect + , Client.onReconnect = Nothing + } + , Client.reconnectTo = Client.ReconnectToOriginal + , Client.onReconnect = def + } | otherwise - = Client.ReconnectAfter def $ do + = Client.ReconnectPolicy $ do threadDelay 10000 - return $ go (n + 1) + return $ + Client.DoReconnect Client.Reconnect { + reconnectTo = Client.ReconnectToOriginal + , onReconnect = def + , nextPolicy = go (n + 1) + } connParams :: Client.ConnParams connParams = def { Client.connReconnectPolicy = reconnectPolicy } @@ -411,4 +425,3 @@ type RPC1 = Trivial' "rpc1" type RPC2 = Trivial' "rpc2" - diff --git a/test-grapesy/Test/Sanity/NoIsLabel.hs b/test-grapesy/Test/Sanity/NoIsLabel.hs new file mode 100644 --- /dev/null +++ b/test-grapesy/Test/Sanity/NoIsLabel.hs @@ -0,0 +1,40 @@ +{-# LANGUAGE OverloadedLabels #-} +{-# OPTIONS_GHC -Wno-orphans #-} + +module Test.Sanity.NoIsLabel (tests) where + +import Data.Proxy +import Data.String +import GHC.OverloadedLabels +import GHC.TypeLits +import Test.Tasty +import Test.Tasty.HUnit + +import Network.GRPC.Common.Protobuf () +import Network.GRPC.Common.Protobuf.Any () + +-- | Too-polymorphic instance of 'IsLabel' +-- +-- \"Good\" 'IsLabel' instances should have a concrete type, rather than just a +-- variable variable @a@. However, some packages (notably @lens@) define a very +-- general instance; this is problematic if @Data.ProtoLens.Labels@ is in scope. +-- We should therefore avoid importing from this module in @grapesy@, leaving +-- the choice whether or not to use @Data.ProtoLens.Labels@ to the user. +-- +-- The point of this test module is to verify that no 'IsLabel' instance is in +-- scope even if we import from @Network.GRPC.Common.Protobuf@. +instance (KnownSymbol symb, IsString q) => IsLabel symb (p -> q) where + fromLabel = \_ -> fromString (symbolVal (Proxy @symb)) + +tests :: TestTree +tests = testGroup "Test.Sanity.NoIsLabel" [ + testCase "Data.ProtoLens.Labels not in scope" thisShouldCompile + ] + +thisShouldCompile :: Assertion +thisShouldCompile = + assertEqual "" "hi" $ f 5 + where + f :: Int -> String + f = #hi +